app-catalog-felhom.eu/audits/persistence-sweep-2026-08-02/state/wishlist_proof.py

#!/usr/bin/env python3
"""wishlist survives-a-redeploy proof, on REAL user data.

`survive2.py` reported DATA-LOST because `prod.db`'s bytes changed across the redeploy — but that
rule cannot tell "the app modified its own database on boot" (fine; vaultwarden's WAL does the
same) from "the app deleted and recreated it" (data gone). Two things settle it:

  1. the file's INODE — same object, or a new one?
  2. a row the USER created — still queryable after the redeploy, or not?

(2) is the claim that actually matters, so it is the verdict. It is created through wishlist's own
HTTP API, never by writing into the volume by hand.
"""
import hashlib, importlib.util, json, os, re, secrets, shutil, tempfile, time
from pathlib import Path

_spec = importlib.util.spec_from_file_location(
    "cvp", "/opt/sweep/scripts/check-volume-persistence.py")
cvp = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(cvp)

TPL = Path("/opt/sweep/templates/wishlist")
EV = Path("/opt/sweep/evidence/wishlist")


def curl(url, *a):
    return cvp._sh(["curl", "-sS", "--max-time", "25", *a, url], timeout=45)


def stat_db(mounts):
    """(inode, sha256, size) of prod.db, read host-side from the volume itself."""
    for m in mounts:
        if m["target"] == "/usr/src/app/data" and m["source"]:
            f = os.path.join(m["source"], "prod.db")
            if os.path.isfile(f):
                st = os.lstat(f)
                return {"inode": st.st_ino, "size": st.st_size,
                        "sha256": hashlib.sha256(open(f, "rb").read()).hexdigest()}
    return None


def sqlite_rows(mounts, table):
    """Count rows in `table` by reading the volume's DB with a throwaway sqlite container —
    a READ of what the app wrote, never a write into the volume."""
    for m in mounts:
        if m["target"] == "/usr/src/app/data" and m["source"]:
            r = cvp._sh(["docker", "run", "--rm", "-v", f"{m['name']}:/v:ro",
                         "keinos/sqlite3:latest", "sqlite3", "/v/prod.db",
                         f"SELECT COUNT(*) FROM {table};"], timeout=180)
            return (r.stdout or r.stderr or "").strip()
    return "no-db"


def up(base, wait=300, settle=40):
    cvp._sh(base + ["up", "-d"], timeout=1200)
    cids = [c for c in cvp._sh(base + ["ps", "-aq"], timeout=120).stdout.split() if c]
    dl = time.time() + wait
    while time.time() < dl:
        if not any(((cvp._inspect(c) or {}).get("State") or {}).get("Status") in ("created", "restarting")
                   or ((((cvp._inspect(c) or {}).get("State") or {}).get("Health") or {}).get("Status")
                       == "starting") for c in cids):
            break
        time.sleep(10)
    ip = None
    for c in cids:
        for n in ((cvp._inspect(c) or {}).get("NetworkSettings") or {}).get("Networks", {}).values():
            if n.get("IPAddress"):
                ip = n["IPAddress"]
    mounts = []
    for c in cids:
        info = cvp._inspect(c) or {}
        for m in (info.get("Mounts") or []):
            mounts.append({"target": m.get("Destination"), "source": m.get("Source"),
                           "name": m.get("Name")})
    time.sleep(settle)
    return cids, ip, mounts


if __name__ == "__main__":
    EV.mkdir(parents=True, exist_ok=True)
    work = Path(tempfile.mkdtemp(prefix="wl-"))
    shutil.copy(TPL / "docker-compose.yml", work / "docker-compose.yml")
    env = cvp.build_env("wishlist", (TPL / ".felhom.yml").read_text(),
                        (TPL / "docker-compose.yml").read_text())
    (work / ".env").write_text("".join(f"{k}={v}\n" for k, v in sorted(env.items())))
    base = ["docker", "compose", "-p", "wlproof", "--project-directory", str(work),
            "-f", str(work / "docker-compose.yml")]
    cvp._sh(["docker", "network", "create", "traefik-public"], timeout=60)
    log = []
    try:
        cids, ip, mounts = up(base)
        # --- seed REAL user data through wishlist's own API
        email = f"proof-{secrets.token_hex(4)}@felhom.invalid"
        pw = "Proof-" + secrets.token_hex(10)              # never recorded
        r = curl(f"http://{ip}:3000/api/auth/signup", "-X", "POST", "-i",
                 "-H", "Content-Type: application/json",
                 "-d", json.dumps({"email": email, "password": pw, "name": "Proof",
                                   "username": "proofuser"}))
        log.append("signup -> " + ((r.stdout or "?").splitlines() or ["?"])[0].strip())
        time.sleep(20)

        before_db = stat_db(mounts)
        before_users = sqlite_rows(mounts, "user")
        log.append(f"before: db={before_db} users={before_users}")

        cvp._sh(base + ["down"], timeout=900)             # a REDEPLOY — no -v
        cids, ip, mounts = up(base)
        after_db = stat_db(mounts)
        after_users = sqlite_rows(mounts, "user")
        log.append(f"after : db={after_db} users={after_users}")

        same_inode = bool(before_db and after_db and before_db["inode"] == after_db["inode"])
        rows_kept = (before_users.isdigit() and after_users.isdigit()
                     and int(after_users) >= int(before_users) and int(before_users) > 0)
        res = {"app": "wishlist", "before": before_db, "after": after_db,
               "user_rows_before": before_users, "user_rows_after": after_users,
               "same_inode": same_inode, "user_rows_preserved": rows_kept,
               "verdict": ("SURVIVES" if same_inode and rows_kept
                           else "SURVIVES (inode only, no seeded rows)" if same_inode
                           else "DATA LOST"),
               "log": log}
        (EV / "survive-wishlist-content.json").write_text(json.dumps(res, indent=2, sort_keys=True))
        print(json.dumps({k: v for k, v in res.items() if k != "log"}, indent=2))
    finally:
        cvp._sh(base + ["down", "-v", "--remove-orphans"], timeout=900)
        shutil.rmtree(work, ignore_errors=True)