#!/usr/bin/env python3 """Campaign 10 — Phase B soak runner. Runs on DooPlex (outside the VM it abuses). Shape (brief B1): a set of atomic operations composed into cycles in varying order, with EVERY invariant checked EVERY cycle — drift at iteration 38 is the point. Every atom and every invariant here was proven BY HAND before automation (see the audit §5); nothing in this file asserts a behaviour that was not first observed live. Writes: journal.tsv (one row per invariant check), runner.log, status.txt (heartbeat). Halts cleanly on: disk floor breach, VM gone, I11 breach (secret leak). """ import json, os, random, subprocess, sys, time, traceback import c10lib as L EV = "/mnt/5_hdd/felhom.eu/git/felhom.eu/documentation/tests/campaign10-evidence-2026-07-31" STATE = os.path.join(EV, "state") os.makedirs(STATE, exist_ok=True) JOURNAL = os.path.join(STATE, "journal.tsv") RUNLOG = os.path.join(STATE, "runner.log") STATUS = os.path.join(STATE, "status.txt") APPS = ["rallly", "homebox", "grafana", "papra"] # docker-volume apps DRIVE_APP = "calibre-web" # binds /mnt/felhom-drives/adatok TARGET_DISK, TARGET_VOL = "scsi2", "c10-scratch:311/vm-311-disk-3.qcow2" # mentes = backup target NONTGT_DISK, NONTGT_VOL = "scsi1", "c10-scratch:311/vm-311-disk-2.qcow2" # adatok FLOOR_GB = 120 # halt if /mnt/nvme-1tb free drops below this MAX_CYCLES = int(os.environ.get("C10_CYCLES", "40")) DEADLINE = time.time() + float(os.environ.get("C10_HOURS", "6")) * 3600 CYCLE = 0 VIOL = [] def log(msg): line = "%s %s" % (time.strftime("%H:%M:%S"), msg) print(line, flush=True) with open(RUNLOG, "a") as fh: fh.write(line + "\n") def rec(inv, ok, detail=""): """One journal row per invariant check. Violations do NOT stop the run (brief B2).""" with open(JOURNAL, "a") as fh: fh.write("%d\t%s\t%s\t%s\t%s\n" % (CYCLE, time.strftime("%FT%TZ", time.gmtime()), inv, "PASS" if ok else "VIOLATION", detail[:400])) if not ok: VIOL.append((CYCLE, inv, detail[:300])) log(" !! %s VIOLATION: %s" % (inv, detail[:220])) def heartbeat(phase): with open(STATUS, "w") as fh: fh.write("cycle=%d\nphase=%s\nutc=%s\nviolations=%d\n" % (CYCLE, phase, time.strftime("%FT%TZ", time.gmtime()), len(VIOL))) # ------------------------------------------------------------------ helpers def hub_events(since="3m"): """Hub-side event stream: type + severity, the observable I1/I2 assert on.""" p = subprocess.run(["sudo", "kubectl", "-n", "felhom-system", "logs", "deploy/hub", "--since=" + since], capture_output=True, text=True, timeout=120) out = [] for line in p.stdout.splitlines(): if "Event from c10-soak:" in line: seg = line.split("Event from c10-soak:", 1)[1].strip() typ = seg.split()[0] sev = seg.split("(", 1)[1].split(")", 1)[0] if "(" in seg else "?" out.append((typ, sev, seg)) return out def target_state(): return (L.jparse(L.api("GET", "/api/storage/backup-target")) or {}).get("data", {}) or {} def disks(): return ((L.jparse(L.api("GET", "/api/disks")) or {}).get("data", {}) or {}).get("disks", []) def wait_backup(timeout=900): t0 = time.time() while time.time() - t0 < timeout: d = ((L.jparse(L.api("GET", "/api/backup/status")) or {}).get("data") or {}) if not d.get("running"): return d time.sleep(10) return {"timeout": True} def wait_restore(timeout=900): t0 = time.time() while time.time() - t0 < timeout: d = ((L.jparse(L.api("GET", "/api/backup/restore-status")) or {}).get("data") or {}) if d and not d.get("running"): return d time.sleep(10) return {"timeout": True} def settle(seconds=130): """Agent tick + host-report + hub event propagation. Measured at ~25-60s; 130 gives margin.""" time.sleep(seconds) # ------------------------------------------------------------------ atoms def atom_backup(): L.api("POST", "/api/backup/run", {}) d = wait_backup() ok = bool(d.get("db_dump", {}).get("success", True)) and not d.get("timeout") rec("BACKUP", ok, json.dumps(d)[:300]) return ok def atom_restore_verify(): """I7 — the whole point: a restore must return the discriminator asked for.""" app = "rallly" # the DB app; its canary is a real DB row genA = "C10-C%03d-A-%s" % (CYCLE, time.strftime("%H%M%S")) L.seed(genA) if not atom_backup(): return genB = "C10-C%03d-B-%s" % (CYCLE, time.strftime("%H%M%S")) L.seed(genB) # must NOT survive the restore snaps = L.jparse(L.api("GET", "/api/backup/snapshots?stack=%s" % app)) or {} pts = snaps.get("data") or [] if not pts: rec("I7", False, "no restore points for %s" % app) return sid = pts[0].get("short_id") t0 = time.time() L.guest('/root/c10api.sh POST /backup/restore "stack_name=%s&snapshot_id=%s" 2>/dev/null' % (app, sid)) d = wait_restore() rto = time.time() - t0 got = L.read_canaries().get(app) rec("I7", got == genA, "app=%s want=%s got=%s snap=%s restore_ok=%s rto=%.1fs" % (app, genA, got, sid, (d.get("last") or {}).get("ok"), rto)) # RTO/RPO byproduct (brief B3): to app-serving-correct-data, not to "restore returned" with open(os.path.join(STATE, "rto.tsv"), "a") as fh: fh.write("%d\t%s\t%s\ttier1\t%.1f\t%s\n" % (CYCLE, time.strftime("%FT%TZ", time.gmtime()), app, rto, got == genA)) def _detach(disk, vol, label, expect_type, expect_recover): """I1/I2 — the absent event must be the RIGHT one, and the pair must match.""" before_boot = L.guest("stat -c %Y /proc/1")[1].strip() L.hp("qm set 311 --delete %s" % disk) settle() evs = hub_events("4m") absent = [e for e in evs if e[0] in ("backup_target_absent", "storage_disconnected")] got = absent[-1][0] if absent else None sev = absent[-1][1] if absent else None rec("I1" if expect_type == "backup_target_absent" else "I2", got == expect_type and sev == "error", "%s absent -> got=%s sev=%s (want %s/error)" % (label, got, sev, expect_type)) # I3: an unusable namespace must not read as bound for d in disks(): if d.get("guest_path", "") and label in str(d.get("guest_path", "")): rec("I3", d.get("bound_under_parent") is False, "%s bound_under_parent=%s while absent" % (label, d.get("bound_under_parent"))) L.hp("qm set 311 --%s %s" % (disk, vol)) settle() evs2 = hub_events("4m") back = [e for e in evs2 if e[0] in ("backup_target_restored", "storage_reconnected")] got2 = back[-1][0] if back else None sev2 = back[-1][1] if back else None rec("I1-pair" if expect_type == "backup_target_absent" else "I2-pair", got2 == expect_recover and sev2 == "info", "%s return -> got=%s sev=%s (want %s/info)" % (label, got2, sev2, expect_recover)) # I4: the bind repairs WITHOUT the guest restarting after_boot = L.guest("stat -c %Y /proc/1")[1].strip() rec("I4", before_boot == after_boot and before_boot != "", "guest init boot before=%s after=%s (must be equal)" % (before_boot, after_boot)) def atom_detach_target(): _detach(TARGET_DISK, TARGET_VOL, "mentes", "backup_target_absent", "backup_target_restored") def atom_detach_nontarget(): _detach(NONTGT_DISK, NONTGT_VOL, "adatok", "storage_disconnected", "storage_reconnected") def atom_reboot_guest(): L.box("pct reboot 9201", timeout=300) for _ in range(40): time.sleep(15) if "felhom-controller" in L.containers(): break rec("REBOOT", "felhom-controller" in L.containers(), "controller back after guest reboot") def atom_kill_controller(): L.guest("docker restart felhom-controller >/dev/null 2>&1 &") time.sleep(45) rec("KILL-CTRL", "felhom-controller" in L.containers(), "controller recovered after kill") def atom_redeploy_app(): app = random.choice(["homebox", "grafana"]) L.api("POST", "/api/stacks/%s/remove" % app, {}) time.sleep(20) sec = L.SECRETS.get(app, {}).get("values", {}) if sec: L.api("POST", "/api/stacks/%s/deploy" % app, {"values": sec}) for _ in range(20): time.sleep(15) if app in L.containers(): break rec("REDEPLOY", app in L.containers(), "%s redeployed" % app) # ------------------------------------------------------------------ per-cycle invariants def check_healthy_baseline(): """I5/I6 — the over-correction check: healthy must render nothing.""" ts = target_state() degraded = ts.get("degraded") evs = [e for e in hub_events("3m") if e[0] in ("backup_target_absent", "storage_disconnected")] rec("I5", degraded is False and not evs, "degraded=%s absent_events=%d target=%s" % (degraded, len(evs), ts.get("target"))) if degraded is False: rec("I6", "message" not in ts or not ts.get("message"), "healthy state must carry no degraded message; got=%s" % str(ts.get("message"))[:120]) UNIT = "/mnt/sys_drive/felhom-data/backups/primary/%s/compose/app.yaml" def check_secret_class_travel(): """I10 — `type: secret` travels to the local unit; `type: password` NEVER does. Asserted in BOTH directions: the withheld class staying withheld is the half that silently rots. Key NAMES only; no value is ever read.""" for app, meta in L.SECRETS.items(): classes = meta.get("secret_classes", {}) if not classes: continue rc, keys, _ = L.guest("grep -oE '^ *[A-Z_0-9]+:' " + (UNIT % app) + " 2>/dev/null | tr -d ' :' | tr '\\n' ' '") present = set(keys.split()) if not present: rec("I10", False, "%s: recovery unit unreadable/absent" % app) continue for ev, c in classes.items(): if c["type"] == "secret": rec("I10", ev in present, "%s/%s type=secret must travel; in_unit=%s" % (app, ev, ev in present)) elif c["type"] == "password": rec("I10", ev not in present, "%s/%s type=password must NEVER travel; in_unit=%s" % (app, ev, ev in present)) def check_secrets_absent(): """I11 — no secret value in any log, event or hub payload. Breach STOPS the run.""" vals = [v for a in L.SECRETS.values() for k, v in a.get("values", {}).items() if k not in ("DOMAIN", "SUBDOMAIN") and v and len(v) > 8] hay = L.guest("docker logs felhom-controller --since 20m 2>&1 | tail -c 200000")[1] p = subprocess.run(["sudo", "kubectl", "-n", "felhom-system", "logs", "deploy/hub", "--since=20m"], capture_output=True, text=True, timeout=120) hay += p.stdout leaked = [v[:6] + "..." for v in vals if v in hay] rec("I11", not leaked, "leaked=%s" % leaked) return not leaked def watchdog(): rc, out, _ = L.hp("df --output=avail -BG /mnt/nvme-1tb | tail -1") try: free = int(out.strip().rstrip("G")) except Exception: free = 9999 rc2, vm, _ = L.hp("qm status 311") alive = "running" in vm if free < FLOOR_GB: log("WATCHDOG: free %dG < floor %dG — halting cleanly" % (free, FLOOR_GB)); return False if not alive: log("WATCHDOG: VM 311 not running (%s) — halting cleanly" % vm.strip()); return False return True # ------------------------------------------------------------------ main ATOMS = [atom_backup, atom_restore_verify, atom_detach_target, atom_detach_nontarget, atom_kill_controller, atom_redeploy_app] def main(): global CYCLE log("=== Campaign 10 Phase B start — max %d cycles, deadline %s ===" % (MAX_CYCLES, time.strftime("%H:%M", time.localtime(DEADLINE)))) if not os.path.exists(JOURNAL): with open(JOURNAL, "w") as fh: fh.write("cycle\tutc\tinvariant\tverdict\tdetail\n") rng = random.Random(20260801) while CYCLE < MAX_CYCLES and time.time() < DEADLINE: CYCLE += 1 heartbeat("start") if not watchdog(): break order = ATOMS[:] rng.shuffle(order) # permutation, not a fixed catalogue # a reboot is expensive; fold it in every 7th cycle only if CYCLE % 7 == 0: order.append(atom_reboot_guest) log("cycle %d: %s" % (CYCLE, " -> ".join(a.__name__.replace("atom_", "") for a in order))) for a in order: heartbeat(a.__name__) try: a() except Exception as e: rec(a.__name__, False, "EXCEPTION %s" % e) log(" atom %s raised: %s" % (a.__name__, traceback.format_exc().splitlines()[-1])) heartbeat("invariants") try: check_healthy_baseline() check_secret_class_travel() if not check_secrets_absent(): log("I11 BREACH — stopping the run immediately per brief B2"); break except Exception as e: log("invariant pass raised: %s" % e) log("cycle %d done — violations so far: %d" % (CYCLE, len(VIOL))) heartbeat("finished") log("=== finished after %d cycles, %d violations ===" % (CYCLE, len(VIOL))) for c, i, d in VIOL: log(" cycle %d %s: %s" % (c, i, d)) if __name__ == "__main__": main()