felhom.eu/documentation/audits/r649-2026-09-23/spike.py

#!/usr/bin/env python3
"""spike.py — Part 1 of the 2026-09-23 brief: the AUTOMATIC UNDO, performed BY HAND on guest 9202.

EVIDENCE, NOT PRODUCT. Every product act goes through the endpoints the UI invokes (deploy, backup,
sync, rescan, update, remove). The UNDO itself has no product path yet (that is what is being
spiked), so it is performed by hand with plain docker/compose inside the guest, using exactly the
steps the product would take, each one timed.

State between stages lives in state-<app>.json so each stage can be run, read, and only then
followed by the next (the hand undo needs a person looking at what the previous step left).
"""
import json, os, sys, time, re
sys.path.insert(0, ".")
import walk as w
import fixtures as fx

HERE = os.path.dirname(os.path.abspath(__file__))
FX = {"docmost": fx.Docmost(), "vikunja": fx.Vikunja(), "romm": fx.Romm()}
SUB = {"docmost": "docs", "vikunja": "tasks", "romm": "arcade"}


def st_path(app):
    return os.path.join(HERE, f"state-{os.environ.get('GUEST', '9202')}-{app}.json")


def load(app):
    return json.load(open(st_path(app))) if os.path.exists(st_path(app)) else {}


def save(app, s):
    json.dump(s, open(st_path(app), "w"), indent=2, ensure_ascii=False)


def ts():
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


# ---- a SECOND seed, written AFTER the backup and BEFORE the update. It is the discriminator: only
# the pre-pin safety dump can hold it — the backup tier copy was taken before it existed. So if it
# reads back after the undo, the undo used the safety dump; if only A reads back, it used the tier.
def docmost_seed_b(sub, A):
    jar = "/tmp/dm.jar"
    w.app_curl(sub, "/api/auth/login", "-c", jar, "-H", "Content-Type: application/json",
               data=json.dumps({"email": A["email"], "password": A["pw"]}), method="POST")
    name = "drillB" + os.urandom(3).hex()
    rc, code, out = w.app_curl(sub, "/api/spaces/create", "-b", jar, "-H", "Content-Type: application/json",
                               data=json.dumps({"name": name, "slug": name.lower()}), method="POST")
    w.say(f"    docmost seed B: /api/spaces/create http={code} {out[:160]}")
    return {"space": name} if code in ("200", "201") else None


def docmost_verify_b(sub, A, B):
    jar = "/tmp/dm.jar"
    rc, code, out = w.app_curl(sub, "/api/auth/login", "-c", jar, "-H", "Content-Type: application/json",
                               data=json.dumps({"email": A["email"], "password": A["pw"]}), method="POST")
    if code not in ("200", "201"):
        w.say(f"    docmost B: cannot log in (http={code})"); return False
    rc, code, out = w.app_curl(sub, "/api/spaces", "-b", jar, "-H", "Content-Type: application/json",
                               data="{}", method="POST")
    ok = code in ("200", "201") and B["space"] in out
    neg = "drillBnever" in out
    w.say(f"    docmost B: /api/spaces http={code} seeded-space-listed={ok} (negative control listed={neg})")
    return ok and not neg


def break_edge(app, frm, to, port_from, port_to):
    """The failing edge: a REAL migrating image move, plus — in the DRILL template only — the
    health probe pointed at a port the app does not answer. Both in one drill commit."""
    fy = f"{w.DRILL}/templates/{app}/.felhom.yml"
    f = open(fy).read()
    m = re.search(r"(healthcheck:\n(?:.*\n){0,8}?\s+port: )" + str(port_from) + r"\b", f)
    assert m, "probe port not found"
    f = f[:m.end() - len(str(port_from))] + str(port_to) + f[m.end():]
    open(fy, "w").write(f)
    h = w.drill_bump(app, frm, to)
    return h


def pg_state(container, db, user):
    return w.guest(f"docker exec {container} psql -U {user} -d {db} -Atc \"select count(*) from information_schema.tables where table_schema='public'\" 2>&1; "
                   f"docker exec {container} psql -U {user} -d {db} -Atc \"select name from kysely_migration order by name desc limit 3\" 2>&1; "
                   f"docker exec {container} psql -U {user} -d {db} -Atc \"select count(*) from kysely_migration\" 2>&1")


def romm_seed_b(sub, A):
    """A SECOND RomM user, created by the first (admin) one — written after the backup."""
    jar, tok = FX["romm"]._csrf(w, sub)
    user = "drillb" + os.urandom(3).hex()
    rc, code, body = w.app_curl(sub, "/api/users", "-b", jar, "-H", f"x-csrftoken: {tok}",
                                "-u", f"{A['user']}:{A['pw']}", "-H", "Content-Type: application/json",
                                data=json.dumps({"username": user, "email": f"{user}@gate.invalid",
                                                 "password": "Drill-" + os.urandom(8).hex(), "role": "viewer"}),
                                method="POST")
    w.say(f"    romm seed B: POST /api/users (as the admin) http={code} {body[:120]}")
    return {"user": user} if code in ("200", "201") else None


def romm_verify_b(sub, A, B):
    jar, tok = FX["romm"]._csrf(w, sub)
    rc, code, body = w.app_curl(sub, "/api/users", "-b", jar, "-H", f"x-csrftoken: {tok}",
                                "-u", f"{A['user']}:{A['pw']}")
    ok = code == "200" and B["user"] in body
    neg = "drillbnever" in body
    w.say(f"    romm B: GET /api/users http={code} seeded-user-listed={ok} (negative control listed={neg})")
    return ok and not neg


def tree(paths):
    cmd = "; ".join(f"echo \"{p}: files=$(find {p} -type f 2>/dev/null | wc -l) sum=$(find {p} -type f -exec sha256sum {{}} + 2>/dev/null | sort | sha256sum | cut -c1-16)\"" for p in paths)
    return w.guest(cmd)


def my_state(container="romm-db", db="romm"):
    # The root password is used INSIDE the container from its own env — it never leaves it.
    q = lambda sql: f"docker exec {container} sh -c 'mariadb -uroot -p\"$MYSQL_ROOT_PASSWORD\" -N -e \"{sql}\" {db}' 2>&1 | tr '\\n' ' '"
    return w.guest(f"""echo -n "tables=$({q("select count(*) from information_schema.tables where table_schema=database()")}) "
echo -n "alembic=$({q("select version_num from alembic_version")}) "
echo -n "users=$({q("select count(*) from users")})"
""")


def vik_seed_b(sub, A):
    """A second project, created AFTER the backup — plus a task with a real ATTACHMENT (a file the
    app writes into its files volume), uploaded through the app's own attachment API."""
    tok, why = FX["vikunja"]._token(w, sub, A)
    title = "drillB-" + os.urandom(4).hex()
    rc, code, body = w.app_curl(sub, "/api/v1/projects", "-H", f"Authorization: Bearer {tok}",
                                "-H", "Content-Type: application/json", data=json.dumps({"title": title}), method="PUT")
    if code not in ("200", "201"):
        w.say(f"    vikunja B: project refused {code} {body[:120]}"); return None
    pid = json.loads(body)["id"]
    rc, code, body = w.app_curl(sub, f"/api/v1/projects/{pid}/tasks", "-H", f"Authorization: Bearer {tok}",
                                "-H", "Content-Type: application/json", data=json.dumps({"title": "task-" + title}), method="PUT")
    tid = json.loads(body)["id"] if code in ("200", "201") else None
    content = "drill attachment " + os.urandom(8).hex()
    fn = "/tmp/vik-att.txt"; open(fn, "w").write(content)
    rc, code2, body2 = w.app_curl(sub, f"/api/v1/tasks/{tid}/attachments", "-H", f"Authorization: Bearer {tok}",
                                  "-F", f"files=@{fn}", method="PUT")
    w.say(f"    vikunja seed B: project http=200 task={tid} attachment upload http={code2} {body2[:120]}")
    return {"title": title, "pid": pid, "tid": tid, "att": content}


def vik_verify_b(sub, A, B):
    tok, why = FX["vikunja"]._token(w, sub, A)
    if not tok:
        w.say(f"    vikunja B: cannot log in {why}"); return {"project": False, "attachment": False}
    rc, code, body = w.app_curl(sub, f"/api/v1/projects/{B['pid']}", "-H", f"Authorization: Bearer {tok}")
    proj = code == "200" and B["title"] in body
    rc, code, body = w.app_curl(sub, f"/api/v1/tasks/{B['tid']}/attachments", "-H", f"Authorization: Bearer {tok}")
    att_ok = False
    try:
        atts = json.loads(body)
        if atts:
            aid = atts[0]["id"]
            rc, c3, b3 = w.app_curl(sub, f"/api/v1/tasks/{B['tid']}/attachments/{aid}", "-H", f"Authorization: Bearer {tok}")
            att_ok = c3 == "200" and B["att"] in b3
    except Exception as e:
        w.say(f"    vikunja B: attachments list unreadable http={code} {body[:120]}")
    w.say(f"    vikunja B: project readback={proj} attachment content readback={att_ok}")
    return {"project": proj, "attachment": att_ok}