amber-backend/scripts/verify.py

"""End-to-end access-rule check for the myanime PocketBase schema.

Run against a local `docker compose up --build` instance that has a superuser.
Overridable via env: PB_BASE, PB_ADMIN_EMAIL, PB_ADMIN_PASS.

    docker compose exec pocketbase pocketbase superuser upsert \\
        admin@myanime.local 'Sup3rSecret!123' --dir=/pb_data
    python scripts/verify.py

Creates test users/profiles/rows, then asserts users can't see or touch each
other's data and can't set their own admin-only `nsfwEnabled`. Exits non-zero
if any check fails.
"""
import json, os, sys, urllib.request, urllib.error

BASE = os.environ.get("PB_BASE", "http://localhost:8090")
ADMIN_EMAIL = os.environ.get("PB_ADMIN_EMAIL", "admin@myanime.local")
ADMIN_PASS = os.environ.get("PB_ADMIN_PASS", "Sup3rSecret!123")

def req(method, path, token=None, body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(BASE + path, data=data, method=method)
    r.add_header("Content-Type", "application/json")
    if token:
        r.add_header("Authorization", token)
    try:
        with urllib.request.urlopen(r) as resp:
            return resp.status, json.loads(resp.read() or "{}")
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read() or "{}")

def show(label, status, ok):
    mark = "PASS" if ok else "FAIL"
    print(f"[{mark}] {label} -> HTTP {status}")
    return ok

allpass = True
def check(label, status, ok):
    global allpass
    allpass = show(label, status, ok) and allpass

# --- superuser ---
st, r = req("POST", "/api/collections/_superusers/auth-with-password",
            body={"identity": ADMIN_EMAIL, "password": ADMIN_PASS})
admin = r["token"]

# --- schema assertions ---
st, users = req("GET", "/api/collections/users", admin)
fnames = {f["name"] for f in users["fields"]}
check("users has username + nsfwEnabled", st, {"username", "nsfwEnabled"} <= fnames)
check("users.createRule blocks nsfwEnabled",
      st, "nsfwEnabled:isset = false" in (users["createRule"] or ""))
check("users.updateRule blocks nsfwEnabled",
      st, "nsfwEnabled:isset = false" in (users["updateRule"] or ""))

for name in ("profiles", "watch_state", "watchlist", "prefs"):
    st, c = req("GET", f"/api/collections/{name}", admin)
    rules_owned = all("@request.auth.id" in (c[k] or "") for k in
                      ("listRule","viewRule","createRule","updateRule","deleteRule"))
    check(f"{name} exists with per-owner rules on all 5 verbs", st, rules_owned)

# unique indexes present
st, ws = req("GET", "/api/collections/watch_state", admin)
check("watch_state has unique (profile,itemId) index", st,
      any("idx_watch_state_profile_item" in i for i in ws["indexes"]))

# --- sync fields (issue #11) ---
def field(coll, name):
    return next((f for f in coll["fields"] if f["name"] == name), None)

check("watch_state.updatedAt is a client-set date, not autodate", st,
      (field(ws, "updatedAt") or {}).get("type") == "date")
check("watch_state.updated is a server autodate (pull cursor)", st,
      (field(ws, "updated") or {}).get("type") == "autodate")
check("watch_state has meta json", st,
      (field(ws, "meta") or {}).get("type") == "json")

st, wl = req("GET", "/api/collections/watchlist", admin)
check("watchlist.addedAt is a client-set date, not autodate", st,
      (field(wl, "addedAt") or {}).get("type") == "date")
check("watchlist.updated is a server autodate (pull cursor)", st,
      (field(wl, "updated") or {}).get("type") == "autodate")
check("watchlist has meta json", st, (field(wl, "meta") or {}).get("type") == "json")
check("watchlist has deletedAt tombstone", st,
      (field(wl, "deletedAt") or {}).get("type") == "date")

st, pf = req("GET", "/api/collections/prefs", admin)
check("prefs.updatedAt is a client-set date", st,
      (field(pf, "updatedAt") or {}).get("type") == "date")

# --- encrypted addon config (issue #20) ---
st, ac = req("GET", "/api/collections/addon_config", admin)
rules_owned = all("@request.auth.id" in (ac.get(k) or "") for k in
                  ("listRule","viewRule","createRule","updateRule","deleteRule"))
check("addon_config exists with per-owner rules on all 5 verbs", st, rules_owned)
check("addon_config.blob is a required text field (ciphertext)", st,
      (field(ac, "blob") or {}).get("type") == "text" and (field(ac, "blob") or {}).get("required"))
check("addon_config has salt + kdf text fields", st,
      (field(ac, "salt") or {}).get("type") == "text" and
      (field(ac, "kdf") or {}).get("type") == "text")
check("addon_config.updatedAt is a client-set date", st,
      (field(ac, "updatedAt") or {}).get("type") == "date")
check("addon_config.updated is a server autodate (pull cursor)", st,
      (field(ac, "updated") or {}).get("type") == "autodate")
check("addon_config has unique (profile) index", st,
      any("idx_addon_config_profile" in i for i in ac.get("indexes", [])))

# --- create two users (superuser create; nsfwEnabled set by admin) ---
def make_user(email, uname):
    st, r = req("POST", "/api/collections/users/records", admin, {
        "email": email, "password": "Passw0rd!23", "passwordConfirm": "Passw0rd!23",
        "username": uname, "nsfwEnabled": True,
    })
    return st, r

st, ua = make_user("alice@example.com", "alice")
check("superuser CAN create user + set nsfwEnabled", st, st == 200 and ua.get("nsfwEnabled") is True)
st, ub = make_user("bob@example.com", "bob")
check("second user created", st, st == 200)

def login(email):
    st, r = req("POST", "/api/collections/users/auth-with-password", body={
        "identity": email, "password": "Passw0rd!23"})
    return r.get("token")

alice = login("alice@example.com")
bob = login("bob@example.com")

# --- alice creates a profile + rows ---
st, pa = req("POST", "/api/collections/profiles/records", alice,
             {"user": ua["id"], "name": "Alice Main"})
check("alice creates her own profile", st, st == 200)
CLIENT_TS = "2026-01-02 03:04:05.000Z"
st, wsrow = req("POST", "/api/collections/watch_state/records", alice,
                {"profile": pa["id"], "itemId": "movie:278:0", "itemType": "movie",
                 "position": 120, "duration": 8520, "updatedAt": CLIENT_TS,
                 "meta": {"title": "The Shawshank Redemption", "episode": 0}})
check("alice creates watch_state under her profile", st, st == 200)

# The client's clock must survive the round-trip untouched — LWW is decided on
# it, so a server that restamped it would silently break conflict resolution.
check("watch_state.updatedAt round-trips the client's value", st,
      wsrow.get("updatedAt", "").startswith("2026-01-02 03:04:05"))
check("watch_state.meta round-trips", st,
      (wsrow.get("meta") or {}).get("title") == "The Shawshank Redemption")
check("watch_state.updated (cursor) is stamped by the server", st,
      bool(wsrow.get("updated")) and not wsrow["updated"].startswith("2026-01-02"))

# A second push of the same item must collide, not duplicate.
st, dup = req("POST", "/api/collections/watch_state/records", alice,
              {"profile": pa["id"], "itemId": "movie:278:0", "itemType": "movie",
               "position": 500, "duration": 8520, "updatedAt": CLIENT_TS})
check("duplicate (profile,itemId) watch_state is rejected", st, st == 400)

# --- watchlist tombstone: remove then re-add must reuse the row -------------
st, wlrow = req("POST", "/api/collections/watchlist/records", alice,
                {"profile": pa["id"], "itemId": "movie:278", "itemType": "movie",
                 "addedAt": CLIENT_TS, "meta": {"title": "Shawshank"}})
check("alice creates watchlist row", st, st == 200)
check("watchlist.addedAt round-trips the client's value", st,
      wlrow.get("addedAt", "").startswith("2026-01-02 03:04:05"))
st, r = req("PATCH", f"/api/collections/watchlist/records/{wlrow['id']}", alice,
            {"deletedAt": "2026-01-03 00:00:00.000Z"})
check("watchlist row soft-deletes", st, st == 200 and bool(r.get("deletedAt")))
st, r = req("PATCH", f"/api/collections/watchlist/records/{wlrow['id']}", alice,
            {"deletedAt": None, "addedAt": "2026-01-04 00:00:00.000Z"})
check("re-add clears the tombstone on the same row (no duplicate)", st,
      st == 200 and not r.get("deletedAt") and r["id"] == wlrow["id"])
st, r = req("GET", "/api/collections/watchlist/records", alice)
check("exactly one watchlist row after delete+re-add", st,
      st == 200 and r.get("totalItems") == 1)

# --- addon_config: store ciphertext + round-trip the client clock -----------
st, acrow = req("POST", "/api/collections/addon_config/records", alice,
                {"profile": pa["id"], "blob": "bm9uY2VjaXBoZXJ0YWc=",
                 "salt": "c29tZXNhbHQ=", "kdf": "pbkdf2-sha256-210000",
                 "updatedAt": CLIENT_TS})
check("alice stores an encrypted addon_config blob", st, st == 200)
check("addon_config.updatedAt round-trips the client's value", st,
      acrow.get("updatedAt", "").startswith("2026-01-02 03:04:05"))
# One row per profile — a second create must collide, not duplicate.
st, dup = req("POST", "/api/collections/addon_config/records", alice,
              {"profile": pa["id"], "blob": "b3RoZXI=", "salt": "c2FsdA==",
               "kdf": "pbkdf2-sha256-210000", "updatedAt": CLIENT_TS})
check("duplicate addon_config for a profile is rejected", st, st == 400)

# --- ISOLATION: bob must not see or touch alice's data ---
st, r = req("GET", f"/api/collections/profiles/records/{pa['id']}", bob)
check("bob CANNOT view alice's profile (404)", st, st == 404)
st, r = req("GET", "/api/collections/watch_state/records", bob)
check("bob's watch_state list is empty", st, st == 200 and r.get("totalItems") == 0)
st, r = req("PATCH", f"/api/collections/watch_state/records/{wsrow['id']}", bob,
            {"position": 999})
check("bob CANNOT update alice's watch_state (404)", st, st == 404)
st, r = req("GET", "/api/collections/addon_config/records", bob)
check("bob's addon_config list is empty", st, st == 200 and r.get("totalItems") == 0)
st, r = req("GET", f"/api/collections/addon_config/records/{acrow['id']}", bob)
check("bob CANNOT view alice's addon_config blob (404)", st, st == 404)

# --- bob cannot create a profile owned by alice ---
st, r = req("POST", "/api/collections/profiles/records", bob,
            {"user": ua["id"], "name": "sneaky"})
check("bob CANNOT create a profile owned by alice", st, st in (400, 403, 404))

# --- alice cannot flip her own nsfwEnabled ---
st, r = req("PATCH", f"/api/collections/users/records/{ua['id']}", alice,
            {"nsfwEnabled": False})
check("alice CANNOT change her nsfwEnabled (rule blocks)", st, st in (400, 403, 404))
st, me = req("GET", f"/api/collections/users/records/{ua['id']}", alice)
check("alice's nsfwEnabled still True after attempt", st, me.get("nsfwEnabled") is True)

# --- alice cannot create watch_state under bob's (nonexistent-to-her) profile ---
st, pb = req("POST", "/api/collections/profiles/records", bob, {"user": ub["id"], "name": "Bob Main"})
st, r = req("POST", "/api/collections/watch_state/records", alice,
            {"profile": pb["id"], "itemId": "movie:238:0", "itemType": "movie",
             "updatedAt": CLIENT_TS})
check("alice CANNOT attach a row to bob's profile", st, st in (400, 403, 404))

print("\n" + ("ALL CHECKS PASSED" if allpass else "SOME CHECKS FAILED"))
sys.exit(0 if allpass else 1)