bindery/tests/test_api.py

"""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 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,
        "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",
        "logout_url": "https://auth.example.com/logout",
    }


@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"),
    ],
)
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


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"<svg/>", 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", "<div>a<br>b</div>", "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", "<p>one</p>", "h1", "new", 0)
    store.add_revision(chapter_id, "<p>two</p>", "h2", "recheck")
    store.add_revision(chapter_id, "<p>three</p>", "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",
        "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'<script type="module" src="static/js/main.js">' in page.data
    assert client.get("/static/js/main.js").status_code == 200
    assert client.get("/static/style.css").status_code == 200


def test_the_page_allows_only_its_own_scripts(client):
    page = client.get("/")
    policy = page.headers["Content-Security-Policy"]
    assert "default-src 'self'" in policy and "unsafe-inline" not in policy
    scripts = re.findall(r"<script>(.*?)</script>", 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