app-catalog-felhom.eu/scripts/upgrade_fixtures.py

#!/usr/bin/env python3
"""Per-app seed/verify fixtures for upgrade-test.py.

THE ONE RULE, carried verbatim from survive2.py: *nothing is ever seeded into a volume by hand.*
R-156's evidence shows a root-written canary making an empty volume read as populated — the exact
confusion this harness exists to remove. So every seed below goes in through the app's OWN
interface: its HTTP API, or its own CLI running inside its own container.

A raw SQL INSERT or a planted file is NOT such a route and is never used. If an app has no
non-browser route, its fixture returns None and the edge is recorded `inconclusive — no non-browser
seed route`, with what was tried. **That is a result**: it tells us which apps can never be
auto-verified, which is a fact nobody currently has.

Each fixture returns an opaque `seeded` token from seed() and answers verify() with it. verify()
must ask the APP, never the filesystem: a migration is supposed to rewrite files.
"""
import base64, json, re, secrets, subprocess, time


def _sh(args, timeout=120, inp=None):
    try:
        return subprocess.run(args, capture_output=True, text=True, timeout=timeout, input=inp)
    except (subprocess.TimeoutExpired, OSError) as e:
        return subprocess.CompletedProcess(args, 124, "", f"{e}")


def _curl(url, *extra, timeout=60, data=None, method=None):
    """One HTTP call. Gates on curl's OWN exit code, never on a summarising pipeline — the trap
    check-image-resolvable.py's docstring names and that has bitten this project twice."""
    args = ["curl", "-sS", "--max-time", str(timeout), "-w", "\n%{http_code}"]
    if method:
        args += ["-X", method]
    if data is not None:
        args += ["--data-binary", "@-"]
    args += list(extra) + [url]
    r = _sh(args, timeout=timeout + 20, inp=data)
    body, _, code = (r.stdout or "").rpartition("\n")
    return r.returncode, code.strip(), body


def _wait_http(url, want, tries=40, delay=5, say=print):
    """Settling says the container is running; this says the APP is answering. They are not the
    same thing, and conflating them is how "the container started" gets reported as a pass."""
    for i in range(tries):
        rc, code, _ = _curl(url, timeout=15)
        if rc == 0 and code in want:
            return True
        time.sleep(delay)
    say(f"    app never answered on {url} (last rc={rc} code={code})")
    return False


# ---------------------------------------------------------------------------------------------
class PrivateBin:
    """PrivateBin's own JSON API. A paste is an HTTP POST and reading it back is an HTTP GET —
    the app stores the blob and hands it back, which is an application-level round trip.

    PrivateBin is FILE-BACKED with no database, so this single seed IS the file half; there is no
    database half to seed separately.
    """
    port = 8080
    container = "privatebin"

    def _base(self, ipfn):
        ip = ipfn(self.container)
        return f"http://{ip}:{self.port}/" if ip else ""

    def seed(self, ipfn, say):
        base = self._base(ipfn)
        if not base:
            say("    privatebin: no container IP")
            return None
        if not _wait_http(base, {"200"}, say=say):
            return None
        marker = "upg-" + secrets.token_hex(8)
        # The v2 paste envelope. PrivateBin validates it server-side: `ct`, the IV and the salt must
        # all be real base64 or the API answers {"status":1,"message":"Invalid data."} — measured,
        # and the reason the first attempt at this fixture failed. The marker is carried INSIDE `ct`
        # so a readback proves THIS paste came back, not merely A paste.
        ct = base64.b64encode(marker.encode()).decode()
        body = json.dumps({
            "v": 2,
            "adata": [[base64.b64encode(secrets.token_bytes(16)).decode(),
                       base64.b64encode(secrets.token_bytes(8)).decode(),
                       100000, 256, 128, "aes", "gcm", "none"], "plaintext", 0, 0],
            "ct": ct,
            "meta": {"expire": "never"},
        })
        rc, code, out = _curl(base, "-H", "X-Requested-With: JSONHttpRequest",
                              "-H", "Content-Type: application/json", data=body, method="POST")
        if rc != 0:
            say(f"    privatebin: POST failed rc={rc}")
            return None
        try:
            j = json.loads(out)
        except Exception:
            say(f"    privatebin: POST returned non-JSON (http {code}): {out[:200]}")
            return None
        if j.get("status") != 0 or not j.get("id"):
            say(f"    privatebin: POST refused: {out[:250]}")
            return None
        say(f"    privatebin: seeded paste id={j['id']}")
        return {"id": j["id"], "marker": ct}

    def verify(self, ipfn, seeded, say):
        base = self._base(ipfn)
        if not base:
            return False
        if not _wait_http(base, {"200"}, tries=24, say=say):
            return False
        rc, code, out = _curl(base + "?pasteid=" + seeded["id"],
                              "-H", "X-Requested-With: JSONHttpRequest")
        if rc != 0 or code != "200":
            say(f"    privatebin: readback rc={rc} http={code}")
            return False
        got = seeded["marker"] in out
        say(f"    privatebin: readback http={code} marker_present={got}")
        return got


# ---------------------------------------------------------------------------------------------
class Docmost:
    """Docmost's own REST API: create the first workspace+user, then a page, then read it back."""
    port = 3000
    container = "docmost"

    def _base(self, ipfn):
        ip = ipfn(self.container)
        return f"http://{ip}:{self.port}" if ip else ""

    def seed(self, ipfn, say):
        base = self._base(ipfn)
        if not base:
            say("    docmost: no container IP")
            return None
        if not _wait_http(base + "/api/health", {"200", "404", "401"}, say=say):
            if not _wait_http(base + "/", {"200", "302", "404"}, tries=20, say=say):
                return None
        marker = "upg-" + secrets.token_hex(8)
        email = f"spike-{secrets.token_hex(4)}@gate.invalid"
        pw = "Spike-" + secrets.token_hex(10)
        setup = json.dumps({"workspaceName": "spike", "name": "spike",
                            "email": email, "password": pw})
        rc, code, out = _curl(base + "/api/auth/setup", "-H", "Content-Type: application/json",
                              "-D", "/tmp/docmost.hdr", data=setup, method="POST")
        say(f"    docmost: /api/auth/setup http={code} rc={rc}")
        if rc != 0 or code not in ("200", "201"):
            say(f"    docmost: setup refused: {out[:250]}")
            return None
        tok = ""
        try:
            tok = json.loads(out).get("tokens", {}).get("accessToken", "") or json.loads(out).get("accessToken", "")
        except Exception:
            pass
        if not tok:
            m = re.search(r"authToken=([^;]+)", open("/tmp/docmost.hdr").read())
            tok = m.group(1) if m else ""
        if not tok:
            say("    docmost: no auth token in the setup response")
            return None
        return {"marker": marker, "email": email, "pw": pw, "token": tok}

    def verify(self, ipfn, seeded, say):
        """Prove the app still holds the seeded ACCOUNT by asking it to authenticate — its own
        front door, and version-stable across the API churn between 0.25 and 0.95."""
        base = self._base(ipfn)
        if not base:
            return False
        if not _wait_http(base + "/", {"200", "302", "404"}, tries=24, say=say):
            return False
        body = json.dumps({"email": seeded["email"], "password": seeded["pw"]})
        rc, code, out = _curl(base + "/api/auth/login", "-H", "Content-Type: application/json",
                              data=body, method="POST")
        ok = rc == 0 and code in ("200", "201")
        say(f"    docmost: login as the seeded user http={code} ok={ok}")
        if not ok:
            say(f"    docmost: login body {out[:200]}")
        return ok


# ---------------------------------------------------------------------------------------------
class BookStack:
    """BookStack has no API token without a browser, so BOTH the seed and the readback go through
    `php artisan` — BookStack's OWN CLI, running inside its own container against its own
    application code and its own User model. That is categorically different from a raw SQL INSERT
    or a planted file, which is what R-156 forbids.

    WHY NOT THE HTTP LOGIN FORM, which was the first attempt and is the more obvious choice:
    BookStack derives APP_URL from the template as `https://${SUBDOMAIN}.${DOMAIN}`, so it marks its
    session and XSRF cookies **`secure`**. curl over plain http therefore stores neither, sends
    neither, and every login POST comes back **419 Page Expired** — measured, and it looks exactly
    like a wrong password. The container serves no TLS, so there is no http route to a logged-in
    session without changing the app's own configuration, which would be testing a different app.

    WHY THE EXIT CODE IS NOT THE GATE HERE, stated because the standing rule says to use it:
    `bookstack:reset-mfa` asks for interactive confirmation, finds no TTY, and exits **1 in both
    cases** — for a user it FOUND and for one it did not. The exit code carries no information, so
    the discriminator is the output, and it is required to be positive AND the not-found sentence is
    required to be ABSENT. It is non-destructive: it aborts at the unanswered prompt.

    THE FIXTURE PROVES ITSELF ON EVERY CALL. Each verify() also probes an email that cannot exist
    and requires the "could not be found" answer. A readback that has broken into always saying
    "found" therefore fails instead of passing everything.

    LIMITATION, recorded rather than papered over: this seeds the DATABASE half only. Seeding a FILE
    (an uploaded image or attachment) needs the API token this app cannot mint headlessly.
    """
    port = 80
    container = "bookstack"

    def _base(self, ipfn):
        ip = ipfn(self.container)
        return f"http://{ip}:{self.port}" if ip else ""

    def _artisan(self, *args, timeout=180):
        for path in ("/app/www/artisan", "/var/www/html/artisan"):
            r = _sh(["docker", "exec", self.container, "php", path] + list(args), timeout=timeout)
            out = (r.stdout or "") + (r.stderr or "")
            if "Could not open input file" not in out:
                return r
        return r

    def _lookup(self, email):
        """Ask BookStack whether it holds this account. Returns True/False/None(unusable)."""
        r = self._artisan("bookstack:reset-mfa", f"--email={email}")
        out = " ".join(((r.stdout or "") + (r.stderr or "")).split())
        found = f"Email: {email}" in out
        missing = "could not be found" in out
        if found == missing:          # neither, or both — the readback itself is broken
            return None, out
        return found, out

    def seed(self, ipfn, say):
        base = self._base(ipfn)
        if not base:
            say("    bookstack: no container IP")
            return None
        if not _wait_http(base + "/login", {"200"}, tries=60, say=say):
            return None
        email = f"spike-{secrets.token_hex(4)}@gate.invalid"
        pw = "Spike-" + secrets.token_hex(10)
        r = self._artisan("bookstack:create-admin", f"--email={email}",
                          f"--name=spike-{secrets.token_hex(3)}", f"--password={pw}")
        out = " ".join(((r.stdout or "") + (r.stderr or "")).split())
        say(f"    bookstack: artisan create-admin rc={r.returncode} :: {out[:120]}")
        if "successfully created" not in out:
            return None
        return {"email": email, "pw": pw}

    def verify(self, ipfn, seeded, say):
        base = self._base(ipfn)
        if not base:
            return False
        # The app must be SERVING, not merely running — "the container started" is not a pass.
        if not _wait_http(base + "/login", {"200"}, tries=60, say=say):
            say("    bookstack: the app never served /login")
            return False
        # The fixture's own negative control, run every time.
        absent, _ = self._lookup(f"nobody-{secrets.token_hex(6)}@gate.invalid")
        if absent is not False:
            say(f"    bookstack: READBACK IS UNUSABLE — an email that cannot exist did not read as absent ({absent})")
            return False
        found, out = self._lookup(seeded["email"])
        say(f"    bookstack: readback of the seeded account found={found} :: {out[:120]}")
        return found is True


# ---------------------------------------------------------------------------------------------
# Added 2026-09-21 by the update night (R-462's widening). Each of these was written and PROVEN
# box-side first, on guest 9202 through the product's own guarded Update, and then ported here so
# the same edge can be run on the harness venue with its ABORT step. The box-side evidence is
# `felhom.eu/documentation/audits/update-night-2026-09-21/apps/<app>/`.
#
# The port is mechanical and one thing changes: box-side the app is reached through traefik with a
# `Host:` header, here through the container's own IP. The SEED ROUTE is identical, and that is the
# expensive half.
# ---------------------------------------------------------------------------------------------


class ActualBudget:
    """Actual's own bootstrap API sets the server password; its own login proves it survived.

    Actual keeps its data in SQLite inside its own volume and performs its own schema migration on
    start, so this single seed is the whole data half.
    """
    port = 5006
    container = "actualbudget"

    def _base(self, ipfn):
        ip = ipfn(self.container)
        return f"http://{ip}:{self.port}" if ip else ""

    def seed(self, ipfn, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/", {"200", "302"}, say=say):
            say("    actualbudget: no container IP, or the app never answered")
            return None
        pw = "Spike-" + secrets.token_hex(10)
        rc, code, out = _curl(base + "/account/bootstrap", "-H", "Content-Type: application/json",
                              data=json.dumps({"password": pw}), method="POST")
        say(f"    actualbudget: /account/bootstrap http={code} :: {out[:140]}")
        if rc != 0 or '"status":"ok"' not in out:
            return None
        return {"pw": pw}

    def verify(self, ipfn, seeded, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/", {"200", "302"}, tries=24, say=say):
            return False

        def login(p):
            return _curl(base + "/account/login", "-H", "Content-Type: application/json",
                         data=json.dumps({"loginMethod": "password", "password": p}),
                         method="POST")

        # the fixture's own negative control, run on every verify
        rc, code, out = login("wrong-" + secrets.token_hex(6))
        if '"status":"ok"' in out:
            say("    actualbudget: READBACK IS UNUSABLE — a wrong password authenticated")
            return False
        rc, code, out = login(seeded["pw"])
        ok = '"status":"ok"' in out
        say(f"    actualbudget: login with the seeded password http={code} ok={ok}")
        return ok


class Navidrome:
    """Navidrome's own /auth/createAdmin makes the first account; its own /auth/login proves it
    survived. LIMITATION: this is the DATABASE half. Navidrome's other half is the music library on
    the drive, which the harness does not populate."""
    port = 4533
    container = "navidrome"

    def _base(self, ipfn):
        ip = ipfn(self.container)
        return f"http://{ip}:{self.port}" if ip else ""

    def seed(self, ipfn, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/", {"200", "302"}, say=say):
            say("    navidrome: no container IP, or the app never answered")
            return None
        user = "spike" + secrets.token_hex(3)
        pw = "Spike-" + secrets.token_hex(10)
        rc, code, out = _curl(base + "/auth/createAdmin", "-H", "Content-Type: application/json",
                              data=json.dumps({"username": user, "password": pw}), method="POST")
        say(f"    navidrome: createAdmin http={code}")
        if rc != 0 or code not in ("200", "201"):
            say(f"    navidrome: refused {out[:200]}")
            return None
        return {"user": user, "pw": pw}

    def verify(self, ipfn, seeded, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/", {"200", "302"}, tries=24, say=say):
            return False

        def login(p):
            return _curl(base + "/auth/login", "-H", "Content-Type: application/json",
                         data=json.dumps({"username": seeded["user"], "password": p}),
                         method="POST")

        rc, code, _ = login("wrong-" + secrets.token_hex(6))
        if code in ("200", "201"):
            say("    navidrome: READBACK IS UNUSABLE — a wrong password authenticated")
            return False
        rc, code, out = login(seeded["pw"])
        ok = code in ("200", "201")
        say(f"    navidrome: login as the seeded user http={code} ok={ok}")
        return ok


class AudiobookShelf:
    """audiobookshelf's own /init creates the first root account; its own /login proves it
    survived. LIMITATION: the DATABASE half only — the library on the drive is not populated."""
    port = 80
    container = "audiobookshelf"

    def _base(self, ipfn):
        ip = ipfn(self.container)
        return f"http://{ip}:{self.port}" if ip else ""

    def seed(self, ipfn, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/status", {"200"}, say=say):
            say("    audiobookshelf: no container IP, or the app never answered /status")
            return None
        user = "spike" + secrets.token_hex(3)
        pw = "Spike-" + secrets.token_hex(10)
        rc, code, out = _curl(base + "/init", "-H", "Content-Type: application/json",
                              data=json.dumps({"newRoot": {"username": user, "password": pw}}),
                              method="POST")
        say(f"    audiobookshelf: /init http={code}")
        if rc != 0 or code not in ("200", "204"):
            say(f"    audiobookshelf: refused {out[:200]}")
            return None
        return {"user": user, "pw": pw}

    def verify(self, ipfn, seeded, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/status", {"200"}, tries=24, say=say):
            return False

        def login(p):
            return _curl(base + "/login", "-H", "Content-Type: application/json",
                         data=json.dumps({"username": seeded["user"], "password": p}),
                         method="POST")

        rc, code, _ = login("wrong-" + secrets.token_hex(6))
        if code == "200":
            say("    audiobookshelf: READBACK IS UNUSABLE — a wrong password authenticated")
            return False
        rc, code, out = login(seeded["pw"])
        ok = code == "200" and seeded["user"] in out
        say(f"    audiobookshelf: login as the seeded root http={code} ok={ok}")
        return ok


class Vikunja:
    """Vikunja's own REST API: register, log in, create a project, read the project back. Four
    calls, all the app's own front door, and the readback is a real authenticated GET."""
    port = 3456
    container = "vikunja"

    def _base(self, ipfn):
        ip = ipfn(self.container)
        return f"http://{ip}:{self.port}" if ip else ""

    def _token(self, base, seeded, pw=None):
        rc, code, out = _curl(base + "/api/v1/login", "-H", "Content-Type: application/json",
                              data=json.dumps({"username": seeded["user"],
                                               "password": pw or seeded["pw"]}), method="POST")
        if code != "200":
            return None
        try:
            return json.loads(out)["token"]
        except Exception:
            return None

    def seed(self, ipfn, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/api/v1/info", {"200"}, say=say):
            say("    vikunja: no container IP, or the app never answered /api/v1/info")
            return None
        user = "spike" + secrets.token_hex(3)
        pw = "Spike-" + secrets.token_hex(10)
        rc, code, out = _curl(base + "/api/v1/register", "-H", "Content-Type: application/json",
                              data=json.dumps({"username": user, "password": pw,
                                               "email": f"{user}@gate.invalid"}), method="POST")
        say(f"    vikunja: register http={code}")
        if code not in ("200", "201"):
            say(f"    vikunja: refused {out[:200]}")
            return None
        seeded = {"user": user, "pw": pw}
        tok = self._token(base, seeded)
        if not tok:
            say("    vikunja: could not log in after registering")
            return None
        title = "spike-" + secrets.token_hex(5)
        # Vikunja CREATES with PUT, not POST — a POST answers `405 Method Not Allowed`, which
        # reads like a broken fixture and is really the wrong verb. Measured 2026-09-21.
        rc, code, out = _curl(base + "/api/v1/projects", "-H", f"Authorization: Bearer {tok}",
                              "-H", "Content-Type: application/json",
                              data=json.dumps({"title": title}), method="PUT")
        say(f"    vikunja: create project http={code}")
        if code not in ("200", "201"):
            say(f"    vikunja: project refused {out[:200]}")
            return None
        seeded["title"] = title
        seeded["pid"] = json.loads(out).get("id")
        return seeded

    def verify(self, ipfn, seeded, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/api/v1/info", {"200"}, tries=24, say=say):
            return False
        if self._token(base, seeded, pw="wrong-" + secrets.token_hex(6)):
            say("    vikunja: READBACK IS UNUSABLE — a wrong password authenticated")
            return False
        tok = self._token(base, seeded)
        if not tok:
            say("    vikunja: the seeded account no longer authenticates")
            return False
        rc, code, out = _curl(base + f"/api/v1/projects/{seeded['pid']}",
                              "-H", f"Authorization: Bearer {tok}")
        ok = code == "200" and seeded["title"] in out
        say(f"    vikunja: readback of the seeded project http={code} ok={ok}")
        return ok


# ---------------------------------------------------------------------------------------------
class Romm:
    """RomM's own user API, driven the way RomM's own front end drives it (added 2026-09-23 for the
    memory watch's red-proof, R-635). Ported from the box-side fixture that walked the 5.0.0 -> 5.3.0
    edge on guest 9202 on 2026-09-21, where each of these was measured rather than guessed:

    - RomM sets a `romm_csrftoken` cookie on any GET and requires it back as an `x-csrftoken`
      header; a bare POST is `403 CSRF token verification failed`, which reads like an auth fault.
    - The fields go in the JSON BODY; a query-string POST is `422 Field required`.
    - The first `POST /api/users` on a fresh install is accepted unauthenticated; afterwards it is
      not — which is what makes `POST /api/login` as that user a real authentication.

    LIMITATION: the DATABASE half only. RomM's other half is the ROM library on the drive.
    """
    port = 8080
    container = "romm"

    def _base(self, ipfn):
        ip = ipfn(self.container)
        return f"http://{ip}:{self.port}" if ip else ""

    def _csrf(self, base):
        jar = f"/tmp/romm-{secrets.token_hex(4)}.jar"
        _curl(base + "/api/heartbeat", "-c", jar)
        tok = ""
        try:
            for line in open(jar):
                if "csrf" in line.lower():
                    tok = line.split()[-1]
        except OSError:
            pass
        return jar, tok

    def seed(self, ipfn, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/api/heartbeat", {"200"}, tries=72, say=say):
            say("    romm: no container IP, or the app never answered /api/heartbeat")
            return None
        jar, tok = self._csrf(base)
        if not tok:
            say("    romm: no romm_csrftoken cookie was set on /api/heartbeat")
            return None
        user = "spike" + secrets.token_hex(3)
        pw = "Spike-" + secrets.token_hex(10)
        rc, code, out = _curl(base + "/api/users", "-b", jar, "-H", f"x-csrftoken: {tok}",
                              "-H", "Content-Type: application/json",
                              data=json.dumps({"username": user, "email": f"{user}@gate.invalid",
                                               "password": pw, "role": "admin"}), method="POST")
        say(f"    romm: POST /api/users http={code}")
        if code not in ("200", "201"):
            say(f"    romm: refused {out[:200]}")
            return None
        return {"user": user, "pw": pw}

    def verify(self, ipfn, seeded, say):
        base = self._base(ipfn)
        if not base or not _wait_http(base + "/api/heartbeat", {"200"}, tries=36, say=say):
            return False
        jar, tok = self._csrf(base)
        rc, code, _ = _curl(base + "/api/login", "-b", jar, "-H", f"x-csrftoken: {tok}",
                            "-u", f"{seeded['user']}:wrong-{secrets.token_hex(5)}", method="POST")
        if code == "200":
            say("    romm: READBACK IS UNUSABLE — a wrong password authenticated")
            return False
        rc, code, out = _curl(base + "/api/login", "-b", jar, "-H", f"x-csrftoken: {tok}",
                              "-u", f"{seeded['user']}:{seeded['pw']}", method="POST")
        ok = code == "200"
        say(f"    romm: login as the seeded user http={code} ok={ok}")
        return ok


FIXTURES = {
    "privatebin": PrivateBin(),
    "docmost": Docmost(),
    "bookstack": BookStack(),
    "actualbudget": ActualBudget(),
    "navidrome": Navidrome(),
    "audiobookshelf": AudiobookShelf(),
    "vikunja": Vikunja(),
    "romm": Romm(),
}