myanime-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"]))

# --- 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)
st, wsrow = req("POST", "/api/collections/watch_state/records", alice,
                {"profile": pa["id"], "itemId": "tt0111161", "itemType": "movie",
                 "position": 120, "duration": 8520})
check("alice creates watch_state under her profile", st, st == 200)

# --- 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)

# --- 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": "tt0068646", "itemType": "movie"})
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)