"""The web API: login modes, the CSRF header, URL_BASE and the web interface.""" import base64 import hashlib import io import re import time import pytest from fake_cwa import FakeCwa from itsdangerous import TimestampSigner from test_build import PNG from werkzeug.security import generate_password_hash from bindery import auth, config, orchestrator, web from bindery.store import Store H = {"X-Requested-With": "bindery"} @pytest.fixture def client(monkeypatch, tmp_path): monkeypatch.setattr(orchestrator, "store", Store(tmp_path / "bindery.db")) monkeypatch.setattr(auth, "limiter", auth.LoginLimiter()) web.app.config["TESTING"] = True return web.app.test_client() @pytest.fixture def cwa(monkeypatch): server = FakeCwa(username="reader", password="pw") monkeypatch.setattr(config, "CWA_URL", server.url) monkeypatch.setattr(config, "AUTH_METHOD", "cwa") yield server server.close() def login(client, username="reader", password="pw", address="127.0.0.1", forwarded=None): headers = {**H, "X-Forwarded-For": forwarded} if forwarded else H return client.post( "/api/auth/login", json={"username": username, "password": password}, headers=headers, environ_base={"REMOTE_ADDR": address}, ) def test_health_and_no_login_mode(client): assert client.get("/api/health").json == {"status": "ok"} check = client.get("/api/auth/check").json assert check["authenticated"] and check["mode"] == "none" assert client.get("/api/stories").json == [] def test_changes_need_the_csrf_header(client): assert client.post("/api/stories", json={"urls": []}).status_code == 403 response = client.post("/api/stories", json={"urls": ["https://example.com/x"]}, headers=H) assert response.status_code == 200 assert response.json["errors"] == ["Not a supported story URL: https://example.com/x"] def test_errors_are_json(client): assert client.get("/api/stories/99").json == {"error": "No story with id 99"} assert client.get("/api/stories/99").status_code == 404 assert client.get("/api/nothing").status_code == 404 def test_cwa_login(client, cwa): assert client.get("/api/stories").status_code == 401 assert client.get("/api/auth/check").json == { "authenticated": False, "mode": "cwa", "user": None, "admin": False, "logout_url": None, } assert login(client, password="wrong").status_code == 401 assert login(client, username="guest", password="").status_code == 401 response = login(client) assert response.status_code == 200 and response.json["user"] == "reader" assert client.get("/api/stories").status_code == 200 client.post("/api/auth/logout", headers=H) assert client.get("/api/stories").status_code == 401 def test_cwa_that_accepts_every_login_cannot_check_a_password(client, cwa, caplog): cwa.anonymous = True for password in ("pw", "wrong"): response = login(client, password=password) assert response.status_code == 401 assert response.json["error"] == "Cannot check the login with CWA. The log has the details." assert "CWA accepts every OPDS login" in caplog.text assert client.get("/api/stories").status_code == 401 # These are not wrong passwords, so they lock no one out. cwa.anonymous = False orchestrator.cwa()._login_test = None assert login(client).status_code == 200 def test_a_session_ends_after_one_day(client, cwa, monkeypatch): response = client.post( "/api/auth/login", json={"username": "reader", "password": "pw", "remember": True}, headers=H ) assert response.status_code == 200 assert client.get("/api/stories").status_code == 200 later = int(time.time()) + 24 * 60 * 60 + 1 monkeypatch.setattr(TimestampSigner, "get_timestamp", lambda self: later) assert client.get("/api/stories").status_code == 401 def test_cwa_login_when_cwa_is_down(client, cwa): cwa.close() assert login(client).json["error"] == "Cannot check the login with CWA. The log has the details." def test_cwa_login_lockout_and_allowed_users(client, cwa, monkeypatch): for _ in range(auth.MAX_FAILURES): login(client, password="wrong") assert login(client).json["error"].startswith("Too many wrong passwords") monkeypatch.setattr(auth, "limiter", auth.LoginLimiter()) monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["someone-else"]) assert login(client).status_code == 401 monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["someone-else", "READER"]) assert login(client).status_code == 200 def test_wrong_passwords_lock_a_name_only_for_their_address(client, cwa): for _ in range(auth.MAX_FAILURES): login(client, password="wrong", address="10.0.0.9") assert login(client, address="10.0.0.9").json["error"].startswith("Too many wrong passwords") assert login(client, address="10.0.0.1").status_code == 200 def test_an_address_has_a_limit_for_all_names(client, cwa): for n in range(auth.MAX_ADDRESS_FAILURES): login(client, username=f"user{n}", password="wrong", address="10.0.0.9") assert login(client, address="10.0.0.9").json["error"].startswith("Too many wrong passwords") assert login(client, address="10.0.0.1").status_code == 200 def test_client_address(monkeypatch): headers = {"X-Forwarded-For": "1.1.1.1, 10.0.0.9"} assert auth.client_address(headers, "172.16.0.2") == "172.16.0.2" monkeypatch.setattr(config, "TRUST_PROXY", True) assert auth.client_address(headers, "172.16.0.2") == "10.0.0.9" assert auth.client_address({}, "172.16.0.2") == "172.16.0.2" def test_only_a_trusted_proxy_sets_the_client_address(client, cwa, monkeypatch): # Without TRUST_PROXY, a client cannot escape the lock with a false header. for n in range(auth.MAX_FAILURES): login(client, password="wrong", address="10.0.0.9", forwarded=f"1.1.1.{n}") assert login(client, address="10.0.0.9", forwarded="1.1.1.9").status_code == 401 # With TRUST_PROXY, clients behind the proxy have their own locks. The proxy adds the last entry. monkeypatch.setattr(config, "TRUST_PROXY", True) monkeypatch.setattr(auth, "limiter", auth.LoginLimiter()) for n in range(auth.MAX_FAILURES): login(client, password="wrong", address="172.16.0.2", forwarded=f"1.1.1.{n}, 10.0.0.9") assert login(client, address="172.16.0.2", forwarded="10.0.0.9").status_code == 401 assert login(client, address="172.16.0.2", forwarded="10.0.0.1").status_code == 200 def test_proxy_header(client, monkeypatch): monkeypatch.setattr(config, "AUTH_METHOD", "proxy") monkeypatch.setattr(config, "PROXY_AUTH_LOGOUT_URL", "https://auth.example.com/logout") assert client.get("/api/stories").status_code == 401 assert client.get("/api/stories", headers={"X-Remote-User": "alice"}).status_code == 200 check = client.get("/api/auth/check", headers={"X-Remote-User": "alice"}).json assert check == { "authenticated": True, "mode": "proxy", "user": "alice", "admin": True, "logout_url": "https://auth.example.com/logout", } @pytest.mark.parametrize(("user", "admin"), [("alice", True), ("Alice", True), ("bob", False)]) def test_only_an_admin_changes_the_global_settings(client, monkeypatch, tmp_path, user, admin): monkeypatch.setattr(config, "PERSONAL_INI", tmp_path / "personal.ini") monkeypatch.setattr(config, "AUTH_METHOD", "proxy") monkeypatch.setattr(config, "ADMIN_USERS", ["alice"]) sid = orchestrator.store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "S", "W", []) headers = {**H, "X-Remote-User": user} status = 200 if admin else 403 assert client.get("/api/auth/check", headers=headers).json["admin"] is admin assert ("personal_ini" in client.get("/api/settings", headers=headers).json) is admin for data in ({"recent_chapters": 5}, {"personal_ini": "[defaults]\n"}): assert client.put("/api/settings", json=data, headers=headers).status_code == status assert (orchestrator.get_settings()["recent_chapters"] == 5) is admin assert config.PERSONAL_INI.exists() is admin rules = {"rules": [{"pattern": "a", "replacement": "b"}]} response = client.put("/api/rules", json=rules, headers=headers) assert response.status_code == status assert bool(orchestrator.store.rules(None)) is admin if not admin: assert response.json == {"error": "Only an admin can do this. BINDERY_ADMIN_USERS lists the admins."} # Each user can read the global rules, and change the rules and the settings of a story. assert client.get("/api/rules", headers=headers).status_code == 200 assert client.put(f"/api/stories/{sid}/rules", json=rules, headers=headers).status_code == 200 assert client.put(f"/api/stories/{sid}/settings", json={"recheck_days": 3}, headers=headers).status_code == 200 def test_the_api_key_and_single_user_modes_are_admins(client, monkeypatch): monkeypatch.setattr(config, "ADMIN_USERS", ["alice"]) assert client.get("/api/auth/check").json["admin"] # the none mode has no users monkeypatch.setattr(config, "AUTH_METHOD", "proxy") monkeypatch.setattr(config, "API_KEY", "secret") key = {"X-Api-Key": "secret"} assert client.get("/api/auth/check", headers=key).json["admin"] assert client.put("/api/settings", json={"recent_chapters": 2}, headers=key).status_code == 200 assert auth.is_admin(auth.Identity("bob", "proxy")) is False monkeypatch.setattr(config, "AUTH_METHOD", "local") assert auth.is_admin(auth.Identity("bob", "session")) # Without a list, each user is an admin. monkeypatch.setattr(config, "AUTH_METHOD", "proxy") monkeypatch.setattr(config, "ADMIN_USERS", []) assert auth.is_admin(auth.Identity("bob", "proxy")) @pytest.mark.parametrize( ("trusted", "address", "accepted"), [ (None, "127.0.0.1", True), (None, "::1", True), (None, "172.18.0.3", True), (None, "192.168.1.20", True), (None, "fd00::5", True), (None, "::ffff:10.1.2.3", True), (None, "203.0.113.9", False), (None, "2001:db8::1", False), (None, "", False), (["172.18.0.3"], "172.18.0.3", True), (["172.18.0.3"], "172.18.0.4", False), (["172.18.0.3"], "127.0.0.1", False), (["10.0.0.0/8", "fd00::/64"], "fd00::7", True), (["10.0.0.0/8", "fd00::/64"], "192.168.1.20", False), ], ) def test_only_a_trusted_address_can_send_the_proxy_header(client, monkeypatch, caplog, trusted, address, accepted): monkeypatch.setattr(config, "AUTH_METHOD", "proxy") if trusted is not None: monkeypatch.setattr(config, "PROXY_AUTH_TRUSTED_IPS", trusted) environ = {"REMOTE_ADDR": address} response = client.get("/api/stories", headers={"X-Remote-User": "alice"}, environ_base=environ) assert response.status_code == (200 if accepted else 401) assert ( client.get("/api/auth/check", headers={"X-Remote-User": "alice"}, environ_base=environ).json["authenticated"] is accepted ) assert ("which is not in PROXY_AUTH_TRUSTED_IPS" in caplog.text) is not accepted # A change needs the header too, from the same address. change = client.post( "/api/stories", json={"urls": []}, headers={**H, "X-Remote-User": "alice"}, environ_base=environ ) assert change.status_code == (200 if accepted else 401) @pytest.mark.parametrize( ("setting", "value", "error"), [ ("AUTH_METHOD", "cwa", "AUTH_METHOD=cwa needs CWA_URL or CWA_DB_PATH"), ("AUTH_METHOD", "cwa-login", "AUTH_METHOD must be none, local, cwa or proxy, not 'cwa-login'"), ("PROXY_AUTH_USER_HEADER", "", "AUTH_METHOD=proxy needs PROXY_AUTH_USER_HEADER"), ("PROXY_AUTH_TRUSTED_IPS", [], "AUTH_METHOD=proxy needs PROXY_AUTH_TRUSTED_IPS, the addresses of the proxy"), ( "PROXY_AUTH_TRUSTED_IPS", ["10.0.0.5", "proxy.local"], "PROXY_AUTH_TRUSTED_IPS must be a list of addresses and networks: " "'proxy.local' does not appear to be an IPv4 or IPv6 network", ), ], ) def test_incomplete_login_settings_refuse_every_request(client, monkeypatch, setting, value, error): monkeypatch.setattr(config, "AUTH_METHOD", "proxy") monkeypatch.setattr(config, "API_KEY", "k" * 32) monkeypatch.setattr(config, setting, value) message = f"{error}. Fix the setting and restart Bindery." for response in ( client.get("/api/auth/check"), client.get("/api/stories", headers={"X-Remote-User": "alice"}), client.get("/api/stories", headers={"X-Api-Key": "k" * 32}), client.post("/api/auth/login", json={"username": "a", "password": "b"}, headers=H), ): assert (response.status_code, response.json) == (503, {"error": message}) assert client.get("/api/health").status_code == 200 assert client.get("/").status_code == 200 @pytest.mark.parametrize( ("mode", "allowed", "host", "accepted"), [ # Without a login, only an address or a name of a private network. ("none", [], "localhost", True), ("none", [], "127.0.0.1:8085", True), ("none", [], "192.168.1.5:8085", True), ("none", [], "[fd00::5]:8085", True), ("none", [], "bindery:8085", True), ("none", [], "nas.local", True), ("none", [], "NAS.Home.Arpa.", True), ("none", [], "bindery.example.com", False), ("none", [], "rebind.attacker.example:8085", False), ("none", [], "notlocal", True), ("none", [], "evil.com.local.attacker.example", False), # With a login, each name. ("local", [], "bindery.example.com", True), ("proxy", [], "bindery.example.com", True), # A list counts in each mode. ("none", ["bindery.example.com"], "bindery.example.com:443", True), ("none", ["bindery.example.com"], "localhost", False), ("none", ["bindery.example.com"], "x.bindery.example.com", False), ("local", [".example.com"], "bindery.example.com", True), ("local", [".example.com"], "example.com", True), ("local", [".example.com"], "badexample.com", False), ("local", ["example.com"], "attacker.example", False), ("none", ["*"], "anything.example", True), ], ) def test_bindery_answers_only_to_its_host_names(client, monkeypatch, mode, allowed, host, accepted): monkeypatch.setattr(config, "AUTH_METHOD", mode) monkeypatch.setattr(config, "LOCAL_USERNAME", "reader") monkeypatch.setattr(config, "LOCAL_PASSWORD_HASH", generate_password_hash("pw", method="pbkdf2:sha256:1000")) monkeypatch.setattr(config, "ALLOWED_HOSTS", allowed) for path in ("/", "/api/auth/check", "/static/style.css"): response = client.get(path, headers={"Host": host}) assert response.status_code == (200 if accepted else 400), path if not accepted: assert response.json == { "error": f"Bindery does not answer to the host name {host}. Add it to ALLOWED_HOSTS." } # The health check of the container works under each name. assert client.get("/api/health", headers={"Host": host}).status_code == 200 def test_api_key(client, cwa, monkeypatch): monkeypatch.setattr(config, "API_KEY", "k" * 32) assert client.get("/api/stories", headers={"X-Api-Key": "wrong"}).status_code == 401 assert client.get("/api/stories", headers={"Authorization": "Bearer " + "k" * 32}).status_code == 200 # A key needs no CSRF header. response = client.post("/api/stories", json={"urls": []}, headers={"X-Api-Key": "k" * 32}) assert response.status_code == 200 def test_epub_upload(client, monkeypatch, tmp_path): calls = [] def queue(path, name, story_id, check): calls.append((path.read_bytes(), name, story_id, check)) return {"id": "job"} monkeypatch.setattr(orchestrator, "queue_epub_file", queue) upload = {"file": (io.BytesIO(b"epub"), "dir/story.epub"), "check": "false"} assert client.post("/api/stories/3/import-epub", data=upload, headers=H).json == {"id": "job"} assert calls == [(b"epub", "story.epub", 3, False)] upload = {"file": (io.BytesIO(b"epub"), "story.epub")} assert client.post("/api/stories/import-epub", data=upload, headers=H).status_code == 200 assert calls[1][2:] == (None, True) missing = client.post("/api/stories/import-epub", data={}, headers=H) assert (missing.status_code, missing.json) == (400, {"error": "Choose an epub file"}) assert client.post("/api/stories/import-epub", data={}).status_code == 403 def test_the_cover_of_a_story(client): store = orchestrator.store story_id = store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", []) assert client.get(f"/api/stories/{story_id}").json["story"]["cover"] is None assert client.get(f"/api/stories/{story_id}/cover").status_code == 404 store.put_image(story_id, "https://example.com/cover.png", "png", "image/png", PNG, is_cover=True) version = client.get(f"/api/stories/{story_id}").json["story"]["cover"] response = client.get(f"/api/stories/{story_id}/cover?v={version}") assert (response.status_code, response.mimetype, response.data) == (200, "image/png", PNG) assert response.cache_control.max_age == 365 * 24 * 60 * 60 assert response.cache_control.private and not response.cache_control.public # A new cover gets a new version, so the browser loads it. store.put_image(story_id, "https://example.com/new.jpg", "jpg", "image/jpeg", b"new", is_cover=True) assert client.get(f"/api/stories/{story_id}").json["story"]["cover"] not in (None, version) # An SVG can hold a script, so Bindery does not serve it. store.put_image(story_id, "https://example.com/cover.svg", "svg", "image/svg+xml", b"", is_cover=True) assert client.get(f"/api/stories/{story_id}").json["story"]["cover"] is None assert client.get(f"/api/stories/{story_id}/cover").status_code == 404 assert client.get("/api/stories/999/cover").status_code == 404 def test_rules_and_the_paragraph_setting(client): story_id = orchestrator.store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", []) chapter_id = orchestrator.store.add_chapter(story_id, "u1", "One", "
a
b
", "h", "new", 0) saved = client.put("/api/rules", json={"rules": [], "paragraphs": True}, headers=H).json assert saved == {"rules": [], "paragraphs": True, "global_paragraphs": True} story = client.get(f"/api/stories/{story_id}/rules").json assert (story["paragraphs"], story["global_paragraphs"]) == (None, True) assert orchestrator._paragraphs(story_id) client.put(f"/api/stories/{story_id}/rules", json={"rules": [], "paragraphs": False}, headers=H) assert not orchestrator._paragraphs(story_id) # Rules without the setting keep it. client.put(f"/api/stories/{story_id}/rules", json={"rules": []}, headers=H) assert client.get(f"/api/stories/{story_id}/rules").json["paragraphs"] is False preview = client.post( f"/api/stories/{story_id}/rules/preview", json={"rules": [], "chapter_id": chapter_id, "source": True}, headers=H, ).json assert preview["identical"] is False # follows the global setting preview = client.post( f"/api/stories/{story_id}/rules/preview", json={"rules": [], "chapter_id": chapter_id, "paragraphs": False}, headers=H, ).json assert preview["identical"] is True def test_delete_revisions(client): store = orchestrator.store story_id = store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", []) chapter_id = store.add_chapter(story_id, "u1", "One", "

one

", "h1", "new", 0) store.add_revision(chapter_id, "

two

", "h2", "recheck") store.add_revision(chapter_id, "

three

", "h3", "recheck") response = client.delete(f"/api/chapters/{chapter_id}/revisions/3", headers=H) assert response.status_code == 400 and "in the book" in response.json["error"] assert [r["rev"] for r in client.delete(f"/api/chapters/{chapter_id}/revisions/2", headers=H).json] == [1, 3] diff = client.get(f"/api/chapters/{chapter_id}/diff").json assert (diff["a"], diff["b"], diff["identical"], diff["story_title"]) == (1, 3, False, "T") assert client.delete(f"/api/chapters/{chapter_id}/revisions/2", headers=H).status_code == 404 # The newest revision stays, also when an older revision is in the book. client.post(f"/api/chapters/{chapter_id}/revision", json={"rev": 1}, headers=H) response = client.delete(f"/api/chapters/{chapter_id}/revisions/3", headers=H) assert response.status_code == 400 and "newest text from the site" in response.json["error"] assert [r["rev"] for r in client.get(f"/api/chapters/{chapter_id}/diff").json["revisions"]] == [1, 3] @pytest.mark.parametrize( "bad", [ {"recent_chapters": "many"}, {"main_jobs_count": 100000}, {"personal_ini": "[defaults\nfoo"}, {"link_books": "yes"}, ], ) def test_a_bad_setting_changes_no_setting(client, monkeypatch, tmp_path, bad): ini = tmp_path / "personal.ini" ini.write_text("[defaults]\n") monkeypatch.setattr(config, "PERSONAL_INI", ini) monkeypatch.setattr(config, "CWA_URL", "https://cwa.invalid") before = client.get("/api/settings").json good = { "link_books": False, "hold_new_links": True, "pause_uploads": True, "recent_chapters": 4, "story_jobs_hours": 1, "personal_ini": "[defaults]\nis_adult:true\n", } assert client.put("/api/settings", json={**good, **bad}, headers=H).status_code == 400 assert client.get("/api/settings").json == before def test_settings(client, monkeypatch, tmp_path): ini = tmp_path / "personal.ini" ini.write_text("[defaults]\nis_adult:true\n") monkeypatch.setattr(config, "PERSONAL_INI", ini) monkeypatch.setattr(config, "CWA_URL", "") monkeypatch.setattr(config, "CALIBRE_DB_PATH", None) assert client.get("/api/settings").json == { "recent_chapters": 0, "link_books": False, "linking_available": False, "hold_new_links": False, "pause_uploads": False, "ingest_available": True, "personal_ini": "[defaults]\nis_adult:true\n", "download_template": "{title} - {authors}", "download_date_format": "%b, %Y", "story_jobs_hours": 36, "story_jobs_count": 10, "main_jobs_hours": 6, "main_jobs_count": 25, } saved = client.put("/api/settings", json={"recent_chapters": 3}, headers=H).json assert (saved["recent_chapters"], saved["warnings"]) == (3, []) saved = client.put("/api/settings", json={"hold_new_links": True, "pause_uploads": True}, headers=H).json assert (saved["hold_new_links"], saved["pause_uploads"], saved["recent_chapters"]) == (True, True, 3) saved = client.put("/api/settings", json={"personal_ini": "[www.notasite.com]\nx:1\n"}, headers=H).json assert saved["warnings"] == ["Line 1: Bad Section Name: [www.notasite.com]"] assert ini.read_text() == "[www.notasite.com]\nx:1\n" bad = client.put("/api/settings", json={"personal_ini": "[defaults\nfoo"}, headers=H) assert bad.status_code == 400 and "line 1" in bad.json["error"] assert ini.read_text() == "[www.notasite.com]\nx:1\n" assert client.put("/api/settings", json={"recent_chapters": -1}, headers=H).status_code == 400 saved = client.put( "/api/settings", json={"story_jobs_hours": 48, "story_jobs_count": 20, "main_jobs_hours": 12, "main_jobs_count": 50}, headers=H, ).json assert (saved["story_jobs_hours"], saved["story_jobs_count"]) == (48, 20) assert (saved["main_jobs_hours"], saved["main_jobs_count"]) == (12, 50) assert client.put("/api/settings", json={"story_jobs_hours": -1}, headers=H).status_code == 400 assert client.put("/api/settings", json={"story_jobs_hours": 721}, headers=H).status_code == 400 assert client.put("/api/settings", json={"main_jobs_count": 501}, headers=H).status_code == 400 def test_metadata_edit(client): story_id = orchestrator.store.create_story( "https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", ["title"] ) response = client.put(f"/api/stories/{story_id}/metadata", json={"values": {"authors": "A & B"}}, headers=H) assert response.json["story"]["pinned"] == {"authors": ["A", "B"]} response = client.put(f"/api/stories/{story_id}/metadata", json={"values": {"title": ""}}, headers=H) assert (response.status_code, response.json) == (400, {"error": "A story needs a title"}) assert client.put(f"/api/stories/{story_id}/metadata", json={}, headers=H).status_code == 400 assert client.put("/api/stories/999/metadata", json={"values": {}}, headers=H).status_code == 404 def test_web_interface_is_served(client): page = client.get("/") assert page.status_code == 200 and b'", page.get_data(as_text=True), re.DOTALL) assert scripts for script in scripts: digest = base64.b64encode(hashlib.sha256(script.encode()).digest()).decode() assert f"'sha256-{digest}'" in policy def test_url_base(): app = web.PrefixMiddleware(web.app.wsgi_app, "/bindery") web.app.wsgi_app, original = app, web.app.wsgi_app try: client = web.app.test_client() assert client.get("/bindery/api/health").status_code == 200 assert client.get("/api/health").status_code == 200 assert client.get("/bindery/static/style.css").status_code == 200 assert client.get("/other").status_code == 404 finally: web.app.wsgi_app = original