"""HTTP-level tests for ani-web (stdlib only), run via pytest. Uses the throwaway server + temp data dir from conftest.py — never touches the real server/data. Seeded users: admin/alice/bob/dave/erin (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", "X-Aniweb": "1"}) 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", "X-Aniweb": "1"}) 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) # ep_id is normalized to int on write (ghost-row fix) 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_history_all_and_delete(base=BASE): jar = login("alice", "alicepw12345", base) e1 = {"ep_id": 201, "slug_id": "del-a", "title": "Del A", "season_id": 1, "ep": "1", "lang": "eng", "pos": 5} e2 = {"ep_id": "202", "slug_id": "del-b", "title": "Del B", "season_id": 1, "ep": "2", "lang": "eng", "pos": 5} assert req("POST", "/api/history", e1, jar=jar, base=base) == 200 assert req("POST", "/api/history", e2, jar=jar, base=base) == 200 all_ = _json("GET", "/api/history?all=1", jar=jar, base=base) assert {e["slug_id"] for e in all_} >= {"del-a", "del-b"} # plain GET still caps at 5 assert len(_json("GET", "/api/history", jar=jar, base=base)) <= 5 # loose str match: the int ep_id row is deleted via a string query param removed = _json("DELETE", "/api/history?slug_id=del-a&ep_id=201", jar=jar, base=base) assert removed["removed"] >= 1, removed all_ = _json("GET", "/api/history?all=1", jar=jar, base=base) assert all(e["slug_id"] != "del-a" for e in all_) assert any(e["slug_id"] == "del-b" for e in all_) # idempotent assert _json("DELETE", "/api/history?slug_id=del-a&ep_id=201", jar=jar, base=base)["removed"] == 0 def test_history_no_ghost_duplicates(base=BASE): """same episode posted with int and string ep_id -> exactly one row""" jar = login("alice", "alicepw12345", base) for ep in (301, "301"): assert req("POST", "/api/history", {"ep_id": ep, "slug_id": "ghost-a", "title": "Ghost", "season_id": 1, "ep": "1", "lang": "eng", "pos": 5}, jar=jar, base=base) == 200 rows = [e for e in _json("GET", "/api/history?all=1", jar=jar, base=base) if e["slug_id"] == "ghost-a"] assert len(rows) == 1 and rows[0]["ep_id"] == 301, rows def test_recent_searches(base=BASE): a = login("alice", "alicepw12345", base) assert req("POST", "/api/searches", {"q": "naruto"}, jar=a, base=base) == 200 assert req("POST", "/api/searches", {"q": "frieren"}, jar=a, base=base) == 200 # case-insensitive dedupe, newest first assert req("POST", "/api/searches", {"q": "NARUTO"}, jar=a, base=base) == 200 assert _json("GET", "/api/searches", jar=a, base=base) == ["NARUTO", "frieren"] # per-user isolation (browser tests write admin's searches, but only after # the API phase, so admin is clean here) b = login("admin", "adminpw12345", base) assert _json("GET", "/api/searches", jar=b, base=base) == [] # blank q rejected assert req("POST", "/api/searches", {"q": " "}, jar=a, base=base) == 422 def test_self_delete(base=BASE): jar = login("dave", "davepw12345", base) # give dave a searches file too: account delete must remove it assert req("POST", "/api/searches", {"q": "dave-search"}, jar=jar, base=base) == 200 assert req("POST", "/account/delete", jar=jar, base=base) == 200 assert "dave" not in json.load(open(_users_file())) assert not os.path.exists(os.path.join(os.environ["ANIWEB_DATA"], "searches", "dave.json")) # old session cookie is dead immediately (middleware re-checks existence) assert req("GET", "/api/me", jar=jar, base=base) == 401 jar2 = CJ() assert req("POST", "/login", {"username": "dave", "password": "davepw12345"}, jar2, base=base) == 401 def test_csrf_header_required(base=BASE): # mutation routes without the X-Aniweb header -> 403 (a cross-site form # POST can't set it; the SPA always sends it). /login is exempt. jar = login("admin", "adminpw12345", base) r = urllib.request.Request(base + "/logout", method="POST") try: assert urllib.request.build_opener( urllib.request.HTTPCookieProcessor(jar)).open(r).status == 403 except urllib.error.HTTPError as e: assert e.code == 403, e.code # with the header (what the SPA sends) it works assert req("POST", "/logout", jar=jar, base=base) == 200 def test_stream_token_host_allowlist(base=BASE): # tokens are plain base64: a crafted token must not let a logged-in user # turn the server into a GET proxy for arbitrary URLs import base64 jar = login("alice", "alicepw12345", base) bad = base64.urlsafe_b64encode(b"https://evil.example/x.m3u8").decode() assert req("GET", f"/stream/{bad}", jar=jar, base=base) == 400 bad2 = base64.urlsafe_b64encode(b"http://hls.anidb.app/x").decode() # non-https assert req("GET", f"/stream/{bad2}", jar=jar, base=base) == 400 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 --- import os # noqa: E402 def _json(method, path, jar=None, base=BASE): r = urllib.request.Request(base + path, method=method, headers={"X-Aniweb": "1"}) 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")