From c697500c7d503b54a36eb9e39b88113bf4c49a6b Mon Sep 17 00:00:00 2001 From: Josue Zamudio Date: Wed, 19 Aug 2026 16:17:56 +0000 Subject: [PATCH] multi-user auth (CLI-created accounts), per-user history, test suite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - server/users.py replaces credentials.py: users.json (0600), PBKDF2, one-time migration of the legacy single password to user admin (history.json -> history/admin.json) - accounts are CLI-managed only: python -m server.users add|rm|list (no self-signup on an internet-exposed app); /signup route removed, login page back to username + password - per-user 30-day signed session cookie (payload carries the username), per-user lockout, GET /api/me, POST /account/delete - per-user watch history: server/data/history/.json; ANIWEB_DATA env override for the data dir - header chip: logged-in user, logout, delete account - tests/: throwaway-server suite (port 8399, temp data dir) — 9 HTTP tests + 2 Playwright browser tests (login UI, real video playback behind auth); runs via python tests/run.py --- README.md | 25 +++++-- SPEC.md | 27 ++++---- server/app.py | 115 +++++++++++++++++++----------- server/credentials.py | 55 --------------- server/users.py | 116 +++++++++++++++++++++++++++++++ static/app.js | 13 ++++ static/index.html | 3 + static/login.html | 23 +++--- static/style.css | 4 ++ tests/run.py | 86 +++++++++++++++++++++++ tests/test_api.py | 158 ++++++++++++++++++++++++++++++++++++++++++ tests/test_browser.py | 92 ++++++++++++++++++++++++ 12 files changed, 595 insertions(+), 122 deletions(-) delete mode 100644 server/credentials.py create mode 100644 server/users.py create mode 100644 tests/run.py create mode 100644 tests/test_api.py create mode 100644 tests/test_browser.py diff --git a/README.md b/README.md index c556eee..eb78c3b 100644 --- a/README.md +++ b/README.md @@ -8,19 +8,30 @@ workflow: search, browse, and stream anime in the browser via anidb.app. ```sh uv venv .venv uv pip install --python .venv/bin/python -r requirements.txt -.venv/bin/python -m server.credentials # set the login password (one-time) +.venv/bin/python -m server.users add # create your account (one-time) ./run.sh ``` Open . `run.sh` restarts cleanly if the server is already running; logs go to `server.log`. The server refuses to -start without a password. +start without at least one user (a legacy `credentials.json` +password, if any, is migrated to user `admin`). + +Account management (no self-signup by design): + +```sh +.venv/bin/python -m server.users add # add user (prompted password) +.venv/bin/python -m server.users rm # remove user + their history +.venv/bin/python -m server.users list # list users +``` ## Features -- Login: single shared password, 30-day session cookie, - 5 wrong attempts → 15-min lockout. PBKDF2-SHA256 hash + session - secret in `server/data/credentials.json` (0600). +- Per-user accounts (CLI-created, no self-signup): 30-day signed + session cookie, 5 wrong attempts → 15-min lockout per user. + PBKDF2-SHA256 hashes in `server/data/users.json` (0600). +- Per-user watch history (`server/data/history/.json`); + in-header chip with logout + delete-account. - Search with poster grid, detail view with synopsis + season tabs - HLS playback (vendored hls.js), browser ABR across 1080/720/360p - Dub/sub toggle @@ -49,6 +60,10 @@ SPEC.md design + the verified anidb.app data flow backend HLS proxy (`/stream/{token}`), which rewrites playlist URIs. - `python server/upstream.py` runs a self-check of the fetchers (search → details → episodes → dub stream for Chainsaw Man). +- `python tests/run.py` runs the full test suite against a throwaway + server (:8399, temp data dir — never touches the real data). Browser + tests need playwright (`pip install playwright && playwright install + chromium`); they're skipped if absent. - Episode titles are not available: anidb's episode API only returns `{id, number, number2, filler}`. Adding titles needs a free MyAnimeList API key. diff --git a/SPEC.md b/SPEC.md index 12a830a..7250df8 100644 --- a/SPEC.md +++ b/SPEC.md @@ -51,19 +51,23 @@ Two moving parts, no database, no framework on the frontend. - `GET /api/anime/{id}/episodes` → `[{id, number, filler}]` (step 3, passthrough) - `GET /api/episode/{ep_id}/stream?lang=eng|jpn` → `{token, qualities}` (steps 5+6; token is base64url of the master m3u8 URL) -- `GET /api/history` / `POST /api/history` → watch history with - position, persisted to `server/data/history.json` (50 kept, top 5 - returned; upsert per anime+episode) +- `GET /api/history` / `POST /api/history` → **per-user** watch + history with position, persisted to + `server/data/history/.json` (50 kept, top 5 returned; + upsert per anime+episode). - `GET /stream/{token}` → proxied HLS. Token = base64url of the upstream URL. Rewrites every non-comment line of m3u8 bodies to point back at `/stream/`; proxies master, variants, and segments (bytes, no transcoding). -- `GET /login` / `POST /login` / `POST /logout` → single-user auth: - PBKDF2-SHA256 password (`server/credentials.py`; hash + session - secret in `server/data/credentials.json`, mode 0600), signed 30-day - cookie, 5-fail/15-min in-memory lockout. Everything else 401s - (API) or redirects to `/login?next=` (pages). Server refuses to - start without credentials. +- `GET /login` / `POST /login` / `POST /logout` → multi-user auth: + PBKDF2-SHA256 passwords (`server/users.py`; `users.json` 0600), + signed 30-day cookie carrying the username, 5-fail/15-min + in-memory lockout **per user**. Accounts are CLI-created only + (`python -m server.users add|rm|list`) — no self-signup. One-time + migration: legacy single-password `credentials.json` → user + `admin`; `history.json` → `history/admin.json`. +- `GET /api/me`, `POST /account/delete` → current user / self-delete + (account + history file). ## Frontend @@ -95,9 +99,8 @@ Open `http://localhost:8321` (bound to 0.0.0.0 for LAN access). ## Out of scope (v1) Downloads (`-d`), next-episode countdown (`-N`), syncplay, intro skip, -ani-schedule integration, multi-user accounts (single shared -password today — per-user history is the natural upgrade), episode -titles (not in +ani-schedule integration, episode titles (not in anidb's API — needs +a MyAnimeList API key). Add when someone asks. anidb's API — needs a MyAnimeList API key). Add when someone asks. ## Risks diff --git a/server/app.py b/server/app.py index 0edbf33..10d01a8 100644 --- a/server/app.py +++ b/server/app.py @@ -6,46 +6,71 @@ Cloudflare workaround) and rewrites playlist URIs to go through itself. """ import base64 import json +import os import time from pathlib import Path from urllib.parse import quote, urljoin -from fastapi import Body, FastAPI, HTTPException, Query +from fastapi import Body, FastAPI, HTTPException, Query, Request from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response from fastapi.staticfiles import StaticFiles from itsdangerous import BadSignature, TimestampSigner -from . import credentials, upstream +from . import upstream, users ROOT = Path(__file__).parent.parent app = FastAPI(title="ani-web") -# ---------- auth: single user, signed cookie ---------- -_SIGNER = TimestampSigner(credentials.load()["secret"]) # exits with instructions if unset +# ---------- auth: multi-user, signed cookies ---------- +users.migrate_legacy() +_SIGNER = TimestampSigner(users.get_secret()) _SESSION = "aniweb_session" _SESSION_AGE = 30 * 86400 _PUBLIC = {("/login", "GET"), ("/login", "POST")} -# ponytail: in-memory lockout, single process — 5 fails lock for 15 min -_lockout = {"fails": 0, "until": 0.0} +# ponytail: in-memory lockout per username, single process — 5 fails -> 15 min +_lockout: dict = {} -def _auth_ok(request) -> bool: +def _session_user(request) -> str | None: try: - return _SIGNER.unsign(request.cookies.get(_SESSION, ""), - max_age=_SESSION_AGE) == b"ok" + payload = _SIGNER.unsign(request.cookies.get(_SESSION, ""), max_age=_SESSION_AGE) except BadSignature: - return False + return None + return payload.decode().split(":", 1)[1] if payload.startswith(b"user:") else None + + +def _lock(user: str, pw: str) -> str: + lo = _lockout.setdefault(user, {"fails": 0, "until": 0.0}) + if time.time() < lo["until"]: + return "locked" + if users.verify(user, pw): + lo.update(fails=0, until=0.0) + return "ok" + lo["fails"] += 1 + if lo["fails"] >= 5: + lo.update(fails=0, until=time.time() + 900) + time.sleep(0.5) # slow down brute force + return "bad" @app.middleware("http") async def require_auth(request, call_next): - if (request.url.path, request.method) in _PUBLIC or _auth_ok(request): + if (request.url.path, request.method) in _PUBLIC: return await call_next(request) - if request.url.path.startswith("/api/"): - return JSONResponse({"detail": "not authenticated"}, status_code=401) - next_ = request.url.path + (f"?{request.url.query}" if request.url.query else "") - return RedirectResponse(f"/login?next={quote(next_, safe='')}", status_code=302) + user = _session_user(request) + if not user: + if request.url.path.startswith("/api/"): + return JSONResponse({"detail": "not authenticated"}, status_code=401) + next_ = request.url.path + (f"?{request.url.query}" if request.url.query else "") + return RedirectResponse(f"/login?next={quote(next_, safe='')}", status_code=302) + request.state.user = user + return await call_next(request) + + +def _session_cookie(resp, user: str): + resp.set_cookie(_SESSION, _SIGNER.sign(b"user:" + user.encode()).decode(), + max_age=_SESSION_AGE, httponly=True, samesite="lax") @app.get("/login", include_in_schema=False) @@ -55,19 +80,30 @@ def login_page(): @app.post("/login") async def login(payload: dict = Body(...)): - if time.time() < _lockout["until"]: - return JSONResponse({"detail": "too many attempts, try later"}, status_code=429) - if credentials.check(payload.get("password", "")): - _lockout.update(fails=0, until=0.0) - resp = JSONResponse({"ok": True}) - resp.set_cookie(_SESSION, _SIGNER.sign(b"ok").decode(), - max_age=_SESSION_AGE, httponly=True, samesite="lax") + u = payload.get("username", "").lower() + if not u: + return JSONResponse({"detail": "username required"}, status_code=400) + state = _lock(u, payload.get("password", "")) + if state == "ok": + resp = JSONResponse({"ok": True, "user": u}) + _session_cookie(resp, u) return resp - _lockout["fails"] += 1 - if _lockout["fails"] >= 5: - _lockout.update(fails=0, until=time.time() + 900) - time.sleep(0.5) # slow down brute force - return JSONResponse({"detail": "wrong password"}, status_code=401) + if state == "locked": + return JSONResponse({"detail": "too many attempts, try later"}, status_code=429) + return JSONResponse({"detail": "wrong username or password"}, status_code=401) + + +@app.get("/api/me", include_in_schema=False) +def api_me(request: Request): + return {"user": request.state.user} + + +@app.post("/account/delete") +def delete_account(request: Request): + users.remove(request.state.user) + resp = JSONResponse({"ok": True}) + resp.delete_cookie(_SESSION) + return resp @app.post("/logout") @@ -77,40 +113,41 @@ def logout(): return resp -HIST_FILE = Path(__file__).parent / "data" / "history.json" +HIST_DIR = Path(os.environ.get("ANIWEB_DATA", Path(__file__).parent / "data")) / "history" -def _load_hist(): +def _load_hist(user: str): try: - return json.loads(HIST_FILE.read_text()) + return json.loads((HIST_DIR / f"{user}.json").read_text()) except (FileNotFoundError, json.JSONDecodeError): return [] -def _save_hist(entries): - HIST_FILE.parent.mkdir(exist_ok=True) - # ponytail: unguarded read-modify-write, fine for a single-user tool - HIST_FILE.write_text(json.dumps(entries, indent=1)) +def _save_hist(user: str, entries): + HIST_DIR.mkdir(exist_ok=True) + # ponytail: unguarded read-modify-write per file, fine at this scale + (HIST_DIR / f"{user}.json").write_text(json.dumps(entries, indent=1)) @app.get("/api/history") -def api_history(): - return sorted(_load_hist(), key=lambda e: e.get("t", 0), reverse=True)[:5] +def api_history(request: Request): + return sorted(_load_hist(request.state.user), key=lambda e: e.get("t", 0), reverse=True)[:5] @app.post("/api/history") -def api_history_post(entry: dict = Body(...)): +def api_history_post(entry: dict = Body(...), request: Request = None): if "ep_id" not in entry: raise HTTPException(422, "ep_id required") entry["t"] = time.time() - entries = _load_hist() + u = request.state.user + entries = _load_hist(u) for i, e in enumerate(entries): if e.get("slug_id") == entry.get("slug_id") and e.get("ep_id") == entry.get("ep_id"): entries[i] = {**e, **entry} break else: entries.append(entry) - _save_hist(sorted(entries, key=lambda e: e.get("t", 0), reverse=True)[:50]) + _save_hist(u, sorted(entries, key=lambda e: e.get("t", 0), reverse=True)[:50]) return entry diff --git a/server/credentials.py b/server/credentials.py deleted file mode 100644 index b06de3f..0000000 --- a/server/credentials.py +++ /dev/null @@ -1,55 +0,0 @@ -"""credentials.py — single-user password + session secret for ani-web. - -Run once: .venv/bin/python -m server.credentials -Stores a PBKDF2 password hash and a random session-signing secret in -server/data/credentials.json (mode 0600). No env vars, no DB. -""" -import getpass -import hashlib -import json -import os -import secrets -from pathlib import Path - -FILE = Path(__file__).parent / "data" / "credentials.json" -ITER = 210_000 # OWASP 2023 minimum for PBKDF2-SHA256 - - -def set_password(): - pw = getpass.getpass("Choose a password (min 8 chars): ") - if len(pw) < 8: - raise SystemExit("password too short") - if pw != getpass.getpass("Again: "): - raise SystemExit("passwords did not match") - salt = secrets.token_bytes(16) - data = { - "v": 1, - "salt": salt.hex(), - "hash": hashlib.pbkdf2_hmac("sha256", pw.encode(), salt, ITER).hex(), - "secret": secrets.token_hex(32), - } - FILE.parent.mkdir(exist_ok=True) - # ponytail: mode 0600 on create; won't tighten an existing file - fd = os.open(FILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) - with os.fdopen(fd, "w") as f: - json.dump(data, f) - print(f"credentials saved to {FILE}") - - -def load(): - if not FILE.exists(): - raise SystemExit( - "no credentials yet — set a password first:\n" - " .venv/bin/python -m server.credentials") - return json.loads(FILE.read_text()) - - -def check(password: str) -> bool: - d = load() - h = hashlib.pbkdf2_hmac("sha256", password.encode(), - bytes.fromhex(d["salt"]), d.get("iter", ITER)) - return secrets.compare_digest(h.hex(), d["hash"]) - - -if __name__ == "__main__": - set_password() diff --git a/server/users.py b/server/users.py new file mode 100644 index 0000000..755064c --- /dev/null +++ b/server/users.py @@ -0,0 +1,116 @@ +"""users.py — multi-user accounts for ani-web (PBKDF2, single json file). + +CLI: .venv/bin/python -m server.users add (password via prompt) + .venv/bin/python -m server.users rm + .venv/bin/python -m server.users list +""" +import getpass +import hashlib +import json +import os +import re +import secrets +import time +from pathlib import Path + +DATA = Path(os.environ.get("ANIWEB_DATA", Path(__file__).parent / "data")) +USERS = DATA / "users.json" +SECRET = DATA / "secret.key" +ITER = 210_000 # OWASP 2023 minimum for PBKDF2-SHA256 +NAME_RE = re.compile(r"^[a-z0-9_-]{3,32}$") + + +def get_secret() -> bytes: + if SECRET.exists(): + return SECRET.read_bytes() + s = secrets.token_bytes(32) + DATA.mkdir(exist_ok=True) + fd = os.open(SECRET, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "wb") as f: + f.write(s) + return s + + +def migrate_legacy(): + """one-time: single-password era -> users.json + per-user history files.""" + cred = DATA / "credentials.json" + if not USERS.exists() and cred.exists(): + d = json.loads(cred.read_text()) + save({"admin": {"salt": d["salt"], "hash": d["hash"], "created": time.time()}}) + cred.unlink() + old_hist = DATA / "history.json" + if not (DATA / "history" / "admin.json").exists() and old_hist.exists(): + (DATA / "history").mkdir(exist_ok=True) + old_hist.rename(DATA / "history" / "admin.json") + + +def load() -> dict: + try: + return json.loads(USERS.read_text()) + except (FileNotFoundError, json.JSONDecodeError): + return {} + + +def save(users: dict): + DATA.mkdir(exist_ok=True) + # ponytail: mode 0600 on create; won't tighten an existing file + fd = os.open(USERS, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as f: + json.dump(users, f, indent=1) + + +def add(username: str, password: str): + u = username.lower() + if not NAME_RE.match(u): + raise ValueError("username: 3-32 chars, only a-z 0-9 _ -") + if len(password) < 8: + raise ValueError("password: min 8 chars") + users = load() + if u in users: + raise ValueError("username already taken") + salt = secrets.token_bytes(16) + users[u] = {"salt": salt.hex(), + "hash": hashlib.pbkdf2_hmac("sha256", password.encode(), salt, ITER).hex(), + "created": time.time()} + save(users) + + +def verify(username: str, password: str) -> bool: + d = load().get(username.lower()) + if not d: + return False + h = hashlib.pbkdf2_hmac("sha256", password.encode(), + bytes.fromhex(d["salt"]), d.get("iter", ITER)) + return secrets.compare_digest(h.hex(), d["hash"]) + + +def remove(username: str): + u = username.lower() + users = load() + users.pop(u, None) + save(users) + (DATA / "history" / f"{u}.json").unlink(missing_ok=True) + + +if __name__ == "__main__": + import sys + cmd = sys.argv[1] if len(sys.argv) > 1 else "" + if cmd == "add" and len(sys.argv) > 2: + name = sys.argv[2] + pw = getpass.getpass(f"Password for {name} (min 8 chars): ") + if pw != getpass.getpass("Again: "): + raise SystemExit("passwords did not match") + try: + add(name, pw) + except ValueError as e: + raise SystemExit(str(e)) + print(f"added user {name.lower()}") + elif cmd == "rm" and len(sys.argv) > 2: + remove(sys.argv[2]) + print(f"removed {sys.argv[2].lower()} (incl. history)") + elif cmd == "list": + for u in sorted(load()): + print(u) + else: + print(__doc__) + raise SystemExit(2) diff --git a/static/app.js b/static/app.js index 253b6aa..3923d6c 100644 --- a/static/app.js +++ b/static/app.js @@ -217,6 +217,19 @@ function showPlayOverlay(v) { // ---------- boot ---------- (async () => { + try { + const me = await api("/api/me"); + document.getElementById("me").textContent = "hi, " + me.user; + const lo = document.getElementById("logout"); + lo.style.display = ""; lo.onclick = async () => { + await fetch("/logout", {method: "POST"}); location.href = "/login"; + }; + const da = document.getElementById("delacct"); + da.style.display = ""; da.onclick = async () => { + if (!confirm("Delete your account and watch history?")) return; + await fetch("/account/delete", {method: "POST"}); location.href = "/login"; + }; + } catch { /* not logged in: api() already redirected to /login */ } await refreshHistory(); const p = new URLSearchParams(location.search); try { diff --git a/static/index.html b/static/index.html index ddf03a4..c50a2bb 100644 --- a/static/index.html +++ b/static/index.html @@ -12,6 +12,9 @@
+ + +

Search for an anime to get started.

diff --git a/static/login.html b/static/login.html index 1902b95..1771c39 100644 --- a/static/login.html +++ b/static/login.html @@ -12,8 +12,8 @@ form { background:var(--card); border:1px solid var(--bd); border-radius:10px; padding:2rem; width:280px; } h1 { font-size:1.1rem; margin:0 0 1rem; color:var(--acc); } input { width:100%; box-sizing:border-box; padding:.55rem .8rem; border-radius:8px; - border:1px solid var(--bd); background:var(--bg); color:var(--fg); } -button { width:100%; margin-top:.8rem; padding:.55rem; border:0; border-radius:8px; + border:1px solid var(--bd); background:var(--bg); color:var(--fg); margin-bottom:.6rem; } +button { width:100%; margin-top:.4rem; padding:.55rem; border:0; border-radius:8px; background:var(--acc); color:#000; font-weight:600; cursor:pointer; } .err { color:#f87171; font-size:.85rem; margin:.6rem 0 0; min-height:1.2em; } @@ -21,22 +21,23 @@ button { width:100%; margin-top:.8rem; padding:.55rem; border:0; border-radius:8

ani-web

- - -

+ + + +

diff --git a/static/style.css b/static/style.css index 2ff5185..34f9282 100644 --- a/static/style.css +++ b/static/style.css @@ -6,6 +6,10 @@ header { display: flex; gap: 1rem; align-items: center; padding: 0.8rem 1.2rem; border-bottom: 1px solid var(--bd); position: sticky; top: 0; background: var(--bg); z-index: 1; } .logo { font-weight: 700; font-size: 1.2rem; color: var(--acc); } header form { flex: 1; max-width: 420px; } +button.tiny { padding: .2rem .55rem; font-size: .8rem; border-radius: 6px; + border: 1px solid var(--bd); background: var(--card); color: var(--fg); cursor: pointer; } +button.tiny.danger { border-color: #7f1d1d; color: #f87171; } +header #me { margin-left: auto; } #q { width: 100%; padding: 0.5rem 0.8rem; border-radius: 8px; border: 1px solid var(--bd); background: var(--card); color: var(--fg); } main { padding: 1.2rem; max-width: 1100px; margin: 0 auto; } diff --git a/tests/run.py b/tests/run.py new file mode 100644 index 0000000..187fe18 --- /dev/null +++ b/tests/run.py @@ -0,0 +1,86 @@ +"""run.py — full ani-web test suite. + +Spins up a throwaway server (127.0.0.1:8399, temp data dir, seeded +users admin/alice/bob), runs the API tests and (if playwright is +installed) the browser tests, then tears everything down. The real +server on :8321 and its data are never touched. + +Usage: .venv/bin/python tests/run.py +""" +import asyncio +import json +import os +import shutil +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.request +from pathlib import Path + +ROOT = Path(__file__).parent.parent +PORT = 8399 +BASE = f"http://127.0.0.1:{PORT}" +USERS = (("admin", "adminpw12345"), ("alice", "alicepw12345"), ("bob", "bobpw12345"), + ("dave", "davepw12345")) + + +def _wait_ready(): + for _ in range(120): + try: + urllib.request.urlopen(BASE + "/login", timeout=1) + return + except Exception: + time.sleep(0.5) + raise SystemExit("server did not start") + + +def main(): + data = Path(tempfile.mkdtemp(prefix="aniweb-test-")) + os.environ["ANIWEB_DATA"] = str(data) + sys.path.insert(0, str(ROOT)) + from server import users + for name, pw in USERS: + users.add(name, pw) + (data / "history").mkdir() + (data / "history" / "admin.json").write_text(json.dumps([ + {"title": "Seed Anime", "slug_id": "seed-1", "season_id": 1, "ep_id": "1", + "lang": "eng", "ep": "1", "pos": 10.0, "t": time.time()}])) + + proc = subprocess.Popen( + [sys.executable, "-m", "uvicorn", "server.app:app", + "--host", "127.0.0.1", "--port", str(PORT)], + cwd=ROOT, env=os.environ, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + _wait_ready() + + import test_api # noqa: E402 (tests/ on sys.path via script dir) + # order matters: lockout test locks bob in memory, run it last + api_tests = [fn for n, fn in sorted(vars(test_api).items()) + if n.startswith("test_") and n != "test_lockout_per_user"] + for fn in api_tests: + print(f"ok api.{fn.__name__}") + fn(BASE) + print("ok api.test_lockout_per_user") + test_api.test_lockout_per_user(BASE) + + try: + import test_browser + except ImportError: + print("SKIP browser tests (pip install playwright && playwright install chromium)") + else: + print("ok browser (playwright)") + asyncio.run(test_browser.test_login_ui()) + asyncio.run(test_browser.test_video_plays_behind_auth()) + + print("ALL TESTS PASSED") + finally: + proc.terminate() + proc.wait() + shutil.rmtree(data, ignore_errors=True) + + +if __name__ == "__main__": + main() diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..8ac9fdb --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,158 @@ +"""HTTP-level tests for ani-web (stdlib only). Run via tests/run.py or +standalone against a running server: python tests/test_api.py + +Uses a throwaway server + temp data dir (see run.py) — never touches +the real server/data. Seeded users: admin/alice/bob (pw pw12345). +""" +import http.cookiejar as _cj +import json +import urllib.error +import urllib.request + +BASE = "http://127.0.0.1:8399" +CJ = lambda: _cj.CookieJar() + + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + """urlopen follows 302s by default; we want to *see* them.""" + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +def no_redirect(method, path, data=None, base=BASE): + """Request without following redirects; returns (status, headers).""" + opener = urllib.request.build_opener(_NoRedirect) + r = urllib.request.Request(base + path, + data=json.dumps(data).encode() if data is not None else None, + method=method, + headers={"Content-Type": "application/json"}) + try: + resp = opener.open(r) + return resp.getcode(), resp.headers + except urllib.error.HTTPError as e: + return e.code, e.headers + + +def no_redirect_get(path, base=BASE): + return no_redirect("GET", path, base=base) + + +def req(method, path, data=None, jar=None, base=BASE): + r = urllib.request.Request(base + path, + data=json.dumps(data).encode() if data is not None else None, + method=method, headers={"Content-Type": "application/json"}) + if jar is not None: + opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) + try: + return opener.open(r).getcode() + except urllib.error.HTTPError as e: + return e.code + try: + return urllib.request.urlopen(r).status + except urllib.error.HTTPError as e: + return e.code + + +def login(user, pw, base=BASE): + jar = CJ() + assert req("POST", "/login", {"username": user, "password": pw}, jar, base) == 200, user + return jar + + +def test_unauthenticated(base=BASE): + # page -> 302 to /login, api -> 401 + code, headers = no_redirect_get("/", base) + assert code == 302 and "/login" in headers["Location"], (code, headers.get("Location")) + assert req("GET", "/api/search?q=x", base=base) == 401 + + +def test_wrong_password(base=BASE): + assert req("POST", "/login", {"username": "admin", "password": "nope"}, base=base) == 401 + + +def test_login_and_access(base=BASE): + jar = login("admin", "adminpw12345", base) + assert req("GET", "/api/search?q=x", jar=jar, base=base) == 200 + assert req("GET", "/api/me", jar=jar, base=base) == 200 + + +def test_logout(base=BASE): + jar = login("admin", "adminpw12345", base) + assert req("POST", "/logout", jar=jar, base=base) == 200 + # session gone: cookie jar emptied by the delete-cookie response + assert not [c for c in jar if c.name == "aniweb_session"], "cookie not cleared" + # and a cookie-less request is bounced to /login + code, headers = no_redirect_get("/", base) + assert code == 302 and "/login" in headers["Location"] + + +def test_no_self_signup(base=BASE): + # /signup route removed — unauthenticated POST bounces to /login + code, headers = no_redirect("POST", "/signup", + {"username": "zed", "password": "whatever123"}, base) + assert code in (302, 404, 405), code + # and even authenticated it's gone (no such route) + jar = login("admin", "adminpw12345", base) + assert req("POST", "/signup", {"username": "zed", "password": "whatever123"}, + jar=jar, base=base) == 405 + assert "zed" not in json.load(open(_users_file())) + + +def test_history_requires_ep_id(base=BASE): + jar = login("admin", "adminpw12345", base) + assert req("POST", "/api/history", {"slug_id": "x"}, jar=jar, base=base) == 422 + + +def test_history_per_user(base=BASE): + a = login("admin", "adminpw12345", base) + b = login("alice", "alicepw12345", base) + entry = {"ep_id": "111", "slug_id": "test-111", "title": "Test", "season_id": 1, + "ep": "111", "lang": "eng", "pos": 42.0} + assert req("POST", "/api/history", entry, jar=a, base=base) == 200 + got_a = _json("GET", "/api/history", jar=a, base=base) + got_b = _json("GET", "/api/history", jar=b, base=base) + assert any(e["ep_id"] == "111" for e in got_a) + assert all(e.get("ep_id") != "111" for e in got_b), "leak: admin entry in alice's history" + + +def test_self_delete(base=BASE): + jar = login("dave", "davepw12345", base) + assert req("POST", "/account/delete", jar=jar, base=base) == 200 + assert "dave" not in json.load(open(_users_file())) + jar2 = CJ() + assert req("POST", "/login", {"username": "dave", "password": "davepw12345"}, jar2, base=base) == 401 + + +def test_lockout_per_user(base=BASE): + # must run on a fresh server (in-memory lockout): 5 fails -> 429 for bob, + # admin unaffected + codes = [req("POST", "/login", {"username": "bob", "password": "bad"}, base=base) + for _ in range(6)] + assert codes[:5] == [401] * 5 and codes[5] == 429, codes + assert req("POST", "/login", {"username": "admin", "password": "adminpw12345"}, base=base) == 200 + + +# --- helpers --- + +def _json(method, path, jar=None, base=BASE): + r = urllib.request.Request(base + path, method=method) + opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) if jar else urllib.request + try: + resp = opener.open(r) + except urllib.error.HTTPError as e: + raise AssertionError(f"{method} {path} -> {e.code}") + return json.loads(resp.read().decode()) + + +def _users_file(): + import os + return os.path.join(os.environ["ANIWEB_DATA"], "users.json") + + +if __name__ == "__main__": + import sys + b = sys.argv[1] if len(sys.argv) > 1 else BASE + for name, fn in sorted(globals().items()): + if name.startswith("test_"): + print("ok", name) + fn(b) diff --git a/tests/test_browser.py b/tests/test_browser.py new file mode 100644 index 0000000..3d89987 --- /dev/null +++ b/tests/test_browser.py @@ -0,0 +1,92 @@ +"""Browser tests for ani-web (Playwright, headless Chromium). + +Run via tests/run.py or standalone: python tests/test_browser.py +Covers what HTTP tests can't: the login UI, the user chip, and real +video playback behind auth. Needs: pip install playwright && playwright install chromium +""" +import asyncio +import sys + +from playwright.async_api import async_playwright + +BASE = "http://127.0.0.1:8399" +LAUNCH_ARGS = ["--autoplay-policy=no-user-gesture-required"] + + +async def _login(page, user, pw): + await page.goto(BASE + "/") + await page.wait_for_selector("input#pw") + assert "create account" not in await page.content(), "self-signup UI must be absent" + await page.fill("#user", user) + await page.fill("#pw", pw) + await page.click("#go") + await page.wait_for_selector("#q", timeout=15000) + await page.wait_for_function( + f"document.getElementById('me').textContent.includes('{user}')", timeout=5000) + + +async def test_login_ui(): + async with async_playwright() as p: + browser = await p.chromium.launch(args=LAUNCH_ARGS) + page = await (await browser.new_context()).new_page() + errors = [] + page.on("pageerror", lambda e: errors.append(str(e))) + + # unauth -> login form + await page.goto(BASE + "/") + await page.wait_for_selector("input#pw") + assert "/login" in page.url, page.url + + # wrong password -> inline error + await page.fill("#user", "admin") + await page.fill("#pw", "wrong") + await page.click("#go") + await page.wait_for_function("document.querySelector('#err').textContent.length > 0", + timeout=10000) + assert "wrong" in await page.inner_text("#err") + + # correct -> home with user chip + await page.fill("#pw", "adminpw12345") + await page.click("#go") + await _login_settled(page) + assert await page.inner_text("#me") == "hi, admin" + assert "Recently played" in await page.inner_text("#main"), "seeded history must show" + assert not errors, errors + await browser.close() + print("ok test_login_ui") + + +async def _login_settled(page): + await page.wait_for_selector("#q", timeout=15000) + await page.wait_for_function( + "document.getElementById('me').textContent.length > 0", timeout=5000) + + +async def test_video_plays_behind_auth(): + async with async_playwright() as p: + browser = await p.chromium.launch(args=LAUNCH_ARGS) + page = await (await browser.new_context()).new_page() + errors = [] + page.on("pageerror", lambda e: errors.append(str(e))) + page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None) + + await _login(page, "alice", "alicepw12345") + await page.fill("#q", "chainsaw") + await page.keyboard.press("Enter") + await page.wait_for_selector(".card", timeout=30000) + await page.click(".card") + await page.wait_for_selector("video", timeout=30000) + await page.wait_for_function( + "document.querySelector('video').readyState >= 4 && document.querySelector('video').currentTime > 1", + timeout=60000) + real_errors = [e for e in errors if "401" not in e] + assert not real_errors, real_errors + await browser.close() + print("ok test_video_plays_behind_auth") + + +if __name__ == "__main__": + if len(sys.argv) > 1: + BASE = sys.argv[1] + asyncio.run(test_login_ui()) + asyncio.run(test_video_plays_behind_auth())