felhom.eu/documentation/audits/probe-fix-2026-09-22/demo.py

#!/usr/bin/env python3
"""demo.py — the guarded Update pressed on the TWO DEMO BOXES, through the product's own endpoints.

Not a harness for the scratch guest: these are the customer-shaped boxes, so this only SYNCS and
presses UPDATE on apps that are already installed. It installs nothing and removes nothing.
"""
import json, os, re, subprocess, sys, time
from datetime import datetime

SC = "/tmp/claude-1000/-mnt-5-hdd-felhom-eu-git/d029e2e6-1762-440e-956d-0760c8aea4b3/scratchpad"
HERE = os.path.dirname(os.path.abspath(__file__))

BOXES = {
    # name        base url            Host header                  ssh host   vmid
    "demo-hp":     (None,                    "felhom.enkisfelhom.hu", "demo-hp",    "9201"),
    "demo-felhom": (None,                    "felhom.demo-felhom.eu", "felhom-pve", "9201"),
}
APPS = ["bookstack", "docmost", "privatebin", "romm"]


def sh(a, timeout=300, inp=None):
    return subprocess.run(a, capture_output=True, text=True, timeout=timeout, input=inp)


class Box:
    def __init__(self, name):
        self.name = name
        base, host, sshh, vmid = BOXES[name]
        self.host, self.ssh, self.vmid = host, sshh, vmid
        self.base = base or self.guest_ip()
        self.sess = self.csrf = None

    def guest_ip(self):
        r = sh(["ssh", "-o", "ConnectTimeout=15", self.ssh,
                f"LC_ALL=C pct exec {self.vmid} -- hostname -I"], timeout=60)
        ip = (r.stdout or "").split()
        if not ip:
            raise SystemExit(f"{self.name}: no guest IP ({r.stderr[:200]})")
        return "https://" + ip[0]

    def guest(self, script):
        r = sh(["ssh", "-o", "ConnectTimeout=15", self.ssh,
                f"LC_ALL=C pct exec {self.vmid} -- bash -s"], timeout=600, inp=script)
        return (r.stdout or "") + (r.stderr or "")

    def login(self):
        pw = open(f"{SC}/.ctlpw").read().strip()
        r = sh(["curl", "-sk", "-D", "-", "-o", "/dev/null", "-H", f"Host: {self.host}",
                "-X", "POST", "--data-urlencode", f"password={pw}", f"{self.base}/login"])
        m = re.search(r"felhom_session=[A-Za-z0-9._-]+", r.stdout or "", re.I)
        if not m:
            raise SystemExit(f"{self.name}: login failed (no session cookie)")
        self.sess = m.group(0)
        r = sh(["curl", "-sk", "-L", "-H", f"Host: {self.host}", "-H", f"Cookie: {self.sess}",
                f"{self.base}/"])
        c = re.search(r'<meta name="csrf-token" content="([^"]+)"', r.stdout or "")
        if not c:
            raise SystemExit(f"{self.name}: login failed (no csrf token)")
        self.csrf = c.group(1)

    def ctl(self, method, path, data=None):
        for attempt in range(2):
            args = ["curl", "-sk", "-H", f"Host: {self.host}", "-H", f"Cookie: {self.sess}",
                    "-w", "\n%{http_code}"]
            if method != "GET":
                args += ["-H", f"X-CSRF-Token: {self.csrf}", "-H", "Content-Type: application/json",
                         "-X", method]
                if data is not None:
                    args += ["--data", json.dumps(data)]
            args.append(f"{self.base}{path}")
            r = sh(args)
            body, _, code = (r.stdout or "").rpartition("\n")
            if code.strip() in ("302", "401") and attempt == 0:
                self.login()
                continue
            try:
                return code.strip(), json.loads(body)
            except Exception:
                return code.strip(), {"_raw": body[:600]}

    def page(self, path):
        r = sh(["curl", "-sk", "-H", f"Host: {self.host}", "-H", f"Cookie: {self.sess}",
                f"{self.base}{path}"])
        return r.stdout or ""

    def stack(self, n):
        _, d = self.ctl("GET", f"/api/stacks/{n}")
        return (d.get("data") or {}) if isinstance(d, dict) else {}


def badge(box, app, lang_suffix=""):
    h = box.page(f"/apps/{app}{lang_suffix}")
    h = re.sub(r"<script.*?</script>|<style.*?</style>", "", h, flags=re.S)
    m = re.findall(r'<span class="tag tag-[^"]*"[^>]*title="([^"]*)"[^>]*>([^<]*)<', h)
    txt = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", h))
    i = txt.find("←")
    return {"tags": [{"title": a.strip(), "text": b.strip()} for a, b in m][:3],
            "context": txt[i:i + 200] if i >= 0 else None}


def observables(box, app):
    st = box.stack(app)
    ac = st.get("app_config") or {}
    comp = box.guest(f"grep -h 'image:' /opt/docker/stacks/{app}/docker-compose.yml 2>/dev/null")
    insp = box.guest("docker ps -a --format '{{.Names}}|{{.Image}}|{{.State}}|{{.Status}}' "
                     f"| grep '^{app}'")
    return {"state": st.get("state"), "pinned_images": ac.get("pinned_images"),
            "installed_images": ac.get("installed_images"),
            "compose": [l.strip() for l in comp.strip().split("\n") if l.strip()],
            "inspect": [l for l in insp.strip().split("\n") if l.strip()]}


def press(box, app, cap_s=1800):
    code, d = box.ctl("POST", f"/api/stacks/{app}/update")
    t0 = time.time()
    phases, last = [], None
    if code != "202":
        return {"accepted": False, "code": code, "answer": str(d)[:300], "phases": []}
    while time.time() - t0 < cap_s:
        st = box.stack(app)
        u = st.get("update") or st.get("update_status") or {}
        ph = u.get("phase") or st.get("update_phase")
        if ph and ph != last:
            phases.append({"t": round(time.time() - t0, 1), "phase": ph,
                           "label": u.get("label") or u.get("message"),
                           "error": u.get("error"), "hold": u.get("hold_message")})
            last = ph
            print("      +%7.1fs  phase=%s  %s" % (phases[-1]["t"], ph, phases[-1]["label"]),
                  flush=True)
            if ph in ("done", "failed"):
                break
        time.sleep(1.0)
    return {"accepted": True, "code": code, "answer": str(d)[:200], "phases": phases}


def main():
    only = sys.argv[1:] or list(BOXES)
    out = {}
    for name in only:
        print(f"\n######## {name}")
        b = Box(name)
        print(f"  base={b.base}  Host={b.host}")
        b.login()
        code, d = b.ctl("GET", "/api/stacks")
        installed = [s["name"] for s in (d.get("data") or []) if s.get("deployed")]
        print("  deployed apps:", ", ".join(sorted(installed)))
        todo = [a for a in APPS if a in installed]
        missing = [a for a in APPS if a not in installed]
        print("  will update:", todo, "| NOT installed here:", missing)
        code, d = b.ctl("POST", "/api/sync")
        print("  sync ->", code, str(d)[:180])
        time.sleep(3)
        b.ctl("POST", "/api/stacks/rescan")
        time.sleep(3)
        rec = {"box": name, "base": b.base, "deployed": sorted(installed),
               "not_installed": missing, "apps": {}}
        for app in todo:
            print(f"\n  ==== {app}")
            before = observables(b, app)
            bhu, ben = badge(b, app), badge(b, app, "?lang=en")
            print("    badge HU:", bhu["tags"])
            print("    badge EN:", ben["tags"])
            print("    before:", before["pinned_images"])
            r = press(b, app)
            after = observables(b, app)
            print("    after :", after["pinned_images"], "| installed:",
                  {k: (v or {}).get("ref") for k, v in (after["installed_images"] or {}).items()})
            phu, pen = badge(b, app), badge(b, app, "?lang=en")
            rec["apps"][app] = {"badge_before_hu": bhu, "badge_before_en": ben,
                                "before": before, "update": r, "after": after,
                                "page_after_hu": phu, "page_after_en": pen}
        out[name] = rec
        json.dump(out, open(os.path.join(HERE, "demo-updates.json"), "w"),
                  ensure_ascii=False, indent=2)
    print("\nwritten demo-updates.json")


if __name__ == "__main__":
    main()