Claude
1d50e5a680
amber-api must know whether an inbound request carries a valid Amber user
token before it spends an ffprobe on a caller-supplied URL. PocketBase
auth tokens are stateless JWTs signed with a per-user secret only
PocketBase holds, so offline verification is impossible by design — the
only correct check is to present the token and see if PocketBase accepts
it. $apis.requireAuth("users") does that before the handler runs, so
reaching the body IS the answer and the body is deliberately trivial.
Returns exactly {id, nsfwEnabled} and nothing else. A token-introspection
endpoint is a tempting place to leak account data (email, username,
profiles) and amber-api has no business knowing any of it — asserted.
Verified against a real PocketBase 0.39.6 running the repo's actual hooks
and migrations (the binary, no Docker needed): full suite green including
8 new assertions — anonymous 401, bogus token 401, valid 200, own id
returned, nsfwEnabled present, no extra keys, per-caller distinctness.
Also verified the integration this exists for: a genuine app token from
PocketBase authenticates a probe enqueue on amber-api, an invalid one
still 401s, and amber-api's 10-minute positive cache means a second
request costs no PocketBase round-trip.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
274 lines
14 KiB
Python
274 lines
14 KiB
Python
"""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", [])))
|
|
|
|
# --- codeless device sign-in (issue #12) ---
|
|
st, da = req("GET", "/api/collections/device_auth", admin)
|
|
check("device_auth collection exists", st, st == 200)
|
|
# Owner-scoped read + delete (the device list + revoke); create/update locked to
|
|
# the hooks (superuser), so those rules are empty.
|
|
check("device_auth list/view/delete are owner-scoped", st,
|
|
all("user = @request.auth.id" in (da.get(k) or "") for k in
|
|
("listRule", "viewRule", "deleteRule")))
|
|
check("device_auth create/update are locked (hooks-only)", st,
|
|
not (da.get("createRule") or "") and not (da.get("updateRule") or ""))
|
|
da_fields = {f["name"]: f for f in da.get("fields", [])}
|
|
check("device_auth has code/status/expiresAt", st,
|
|
{"code", "status", "expiresAt"} <= set(da_fields))
|
|
check("device_auth hides authToken/keyCiphertext/pollSecret/devicePubKey", st,
|
|
all(da_fields.get(n, {}).get("hidden") for n in
|
|
("authToken", "keyCiphertext", "pollSecret", "devicePubKey")))
|
|
check("device_auth.code has a unique index", st,
|
|
any("idx_device_auth_code" in i for i in da.get("indexes", [])))
|
|
|
|
# The custom routes are registered (hooks loaded). A bad request still proves the
|
|
# route exists (400/404, not the 404-with-no-such-route body). The approve page
|
|
# GET /device should return HTML.
|
|
st, _ = req("POST", "/api/device-auth/request", body={"deviceName": "verify.py",
|
|
"devicePubKey": "AAAA"})
|
|
check("POST /api/device-auth/request is served (hook loaded)", st, st == 200)
|
|
st, _ = req("GET", "/api/device-auth/poll?id=nope&secret=nope")
|
|
check("GET /api/device-auth/poll is served", st, st in (200, 400, 403, 404))
|
|
|
|
# --- 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))
|
|
|
|
# --- /api/amber/whoami: token introspection for amber-api ---
|
|
# amber-api cannot verify a PocketBase JWT itself (stateless, signed with a
|
|
# per-user secret PocketBase alone holds), so it asks this route. These checks
|
|
# guard two things: that it fails closed for anyone unauthenticated, and that it
|
|
# leaks nothing beyond the two fields amber-api needs.
|
|
st, r = req("GET", "/api/amber/whoami")
|
|
check("whoami: anonymous → 401 (fails closed)", st, st == 401)
|
|
|
|
st, r = req("GET", "/api/amber/whoami", "Bearer totally-not-a-token")
|
|
check("whoami: bogus token → 401", st, st == 401)
|
|
|
|
st, r = req("GET", "/api/amber/whoami", alice)
|
|
check("whoami: valid token → 200", st, st == 200)
|
|
check("whoami returns the caller's own id", st, r.get("id") == ua["id"])
|
|
check("whoami returns nsfwEnabled", st, r.get("nsfwEnabled") is True)
|
|
# A token-introspection endpoint is a tempting place to leak account data;
|
|
# amber-api has no business knowing the email, username or profile list.
|
|
check("whoami leaks nothing else", st,
|
|
set(r.keys()) == {"id", "nsfwEnabled"})
|
|
|
|
# Per-caller, not "some authenticated user": a shared/cached answer here would
|
|
# let amber-api attribute one device's requests to another account.
|
|
st, r = req("GET", "/api/amber/whoami", bob)
|
|
check("whoami is per-caller (bob's token returns bob's id)", st, r.get("id") == ub["id"])
|
|
check("whoami ids are distinct per caller", st, ua["id"] != ub["id"])
|
|
|
|
print("\n" + ("ALL CHECKS PASSED" if allpass else "SOME CHECKS FAILED"))
|
|
sys.exit(0 if allpass else 1)
|