felhom.eu/documentation/tests/campaign10-evidence-2026-07-31/runner/c10run.py

#!/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()