bindery/tests/test_opds_settings.py

"""The OPDS catalog stays off until an admin enables it in Settings. An admin also sets its password there."""

import base64

import pytest
from test_api import H
from test_api import client as client
from werkzeug.security import check_password_hash, generate_password_hash

from bindery import config, orchestrator, web
from bindery.store import Store

PATHS = ["/opds", "/opds/", "/opds/title", "/opds/author", "/opds/changed"]


def test_catalog_defaults_to_off_and_can_be_toggled_without_restart(client, monkeypatch):
    assert client.get("/api/settings").json["opds_enabled"] is False
    assert orchestrator.store.get_setting(None, "opds_enabled") is None
    for enabled in (False, True, False):
        if enabled or orchestrator.store.get_setting(None, "opds_enabled") is not None:
            saved = client.put("/api/settings", json={"opds_enabled": enabled}, headers=H)
            assert saved.status_code == 200
            assert saved.json["opds_enabled"] is enabled
        for path in PATHS:
            for method in (client.get, client.head):
                assert method(path).status_code == (200 if enabled else 404)
        if enabled:
            # A fresh store instance reads the saved setting, as after a restart.
            monkeypatch.setattr(orchestrator, "store", Store(orchestrator.store.path))
            assert client.get("/api/settings").json["opds_enabled"] is True
            unchanged = client.put("/api/settings", json={"recent_chapters": 2}, headers=H)
            assert unchanged.json["opds_enabled"] is True
            unchanged = client.put("/api/settings", json={"opds_enabled": None}, headers=H)
            assert unchanged.json["opds_enabled"] is True


@pytest.mark.parametrize("value", ["false", "true", 0, 1, [], {}])
def test_opds_setting_requires_a_boolean_and_rejects_all_changes(client, value):
    before = client.get("/api/settings").json
    response = client.put("/api/settings", json={"opds_enabled": value, "recent_chapters": 9}, headers=H)
    assert response.status_code == 400
    assert "true or false" in response.json["error"]
    assert client.get("/api/settings").json == before


def test_invalid_download_settings_do_not_enable_opds(client):
    response = client.put(
        "/api/settings", json={"opds_enabled": True, "download_template": "{title:uppercase()}"}, headers=H
    )
    assert response.status_code == 400
    assert client.get("/api/settings").json["opds_enabled"] is False
    assert client.get("/opds").status_code == 404


def test_only_an_admin_can_enable_opds_and_csrf_header_is_required(client, monkeypatch):
    monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
    monkeypatch.setattr(config, "ADMIN_USERS", ["admin"])
    monkeypatch.setattr(config, "PROXY_AUTH_TRUSTED_IPS", ["127.0.0.1"])
    assert client.put("/api/settings", json={"opds_enabled": True}, headers=H).status_code == 401
    assert (
        client.put("/api/settings", json={"opds_enabled": True}, headers={**H, "X-Remote-User": "reader"}).status_code
        == 403
    )
    assert (
        client.put("/api/settings", json={"opds_enabled": True}, headers={"X-Remote-User": "admin"}).status_code == 403
    )
    assert orchestrator.store.get_setting(None, "opds_enabled") is None
    response = client.put("/api/settings", json={"opds_enabled": True}, headers={**H, "X-Remote-User": "admin"})
    assert response.status_code == 200
    assert response.json["opds_enabled"] is True
    assert client.get("/opds").status_code == 401
    # The proxy user header does not open the catalog. Only the catalog password does.
    assert client.get("/opds", headers={"X-Remote-User": "reader"}).status_code == 401


PASSWORD_URL = "/api/settings/opds-password"


def basic(password, user="reader"):
    return {"Authorization": "Basic " + base64.b64encode(f"{user}:{password}".encode()).decode()}


def as_user(name):
    return {**H, "X-Remote-User": name}


@pytest.fixture
def proxy(monkeypatch):
    monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
    monkeypatch.setattr(config, "PROXY_AUTH_TRUSTED_IPS", ["127.0.0.1"])
    monkeypatch.setattr(config, "ADMIN_USERS", ["admin"])
    monkeypatch.setattr(config, "API_KEY", "key")


def test_disabled_catalog_answers_404_under_url_base_with_any_credentials(client, proxy, monkeypatch):
    monkeypatch.setattr(web.app, "wsgi_app", web.PrefixMiddleware(web.app.wsgi_app, "/bindery"))
    keyed = {"X-Api-Key": "key"}
    for path in PATHS:
        for headers in (keyed, basic("reader-password")):
            assert client.get("/bindery" + path, headers=headers).status_code == 404
    assert client.put("/bindery/api/settings", json={"opds_enabled": True}, headers=keyed).status_code == 200
    saved = client.put("/bindery" + PASSWORD_URL, json={"password": "reader-password"}, headers=as_user("reader"))
    assert saved.status_code == 200
    for path in PATHS:
        assert client.get("/bindery" + path, headers=keyed).status_code == 401
        assert client.get("/bindery" + path, headers=basic("reader-password")).status_code == 200


def test_each_user_sets_their_own_password_and_it_is_stored_as_a_hash(client, proxy):
    client.put("/api/settings", json={"opds_enabled": True}, headers=as_user("admin"))
    for user in ("reader", "admin"):
        info = client.get("/api/settings", headers=as_user(user)).json
        assert (info["opds_user"], info["opds_password_set"]) == (user, False)
    saved = client.put(PASSWORD_URL, json={"password": "first-password"}, headers=as_user("Reader"))
    assert saved.json == {"opds_user": "Reader", "opds_password_set": True}
    stored = orchestrator.store.get_setting(None, "opds_password_hash:reader")
    assert "first-password" not in stored
    assert check_password_hash(stored, "first-password")
    # The answers show the user name and the state, never the password or its hash.
    for response in (saved, client.get("/api/settings", headers=as_user("reader"))):
        assert "first-password" not in response.text
        assert stored not in response.text
    assert client.get("/api/settings", headers=as_user("reader")).json["opds_password_set"] is True
    assert client.get("/api/settings", headers=as_user("admin")).json["opds_password_set"] is False
    assert client.get("/opds", headers=basic("first-password", "reader")).status_code == 200
    assert client.get("/opds", headers=basic("first-password", "admin")).status_code == 401
    # A new password replaces the old one. Other changes leave it as it is.
    client.put(PASSWORD_URL, json={"password": "second-password"}, headers=as_user("reader"))
    client.put("/api/settings", json={"recent_chapters": 2}, headers=as_user("admin"))
    assert client.get("/opds", headers=basic("first-password")).status_code == 401
    assert client.get("/opds", headers=basic("second-password")).status_code == 200
    client.put(PASSWORD_URL, json={"password": "admin-password"}, headers=as_user("admin"))
    assert client.get("/opds", headers=basic("admin-password", "admin")).status_code == 200
    assert client.get("/opds", headers=basic("second-password")).status_code == 200


@pytest.mark.parametrize("value", [None, "", "short", "x" * 7, "x" * 257, 12345678, ["password"], True])
def test_bad_password_changes_nothing(client, proxy, value):
    response = client.put(PASSWORD_URL, json={"password": value}, headers=as_user("reader"))
    assert response.status_code == 400
    assert "OPDS password" in response.json["error"]
    assert orchestrator.store.get_setting(None, "opds_password_hash:reader") is None


def test_a_password_of_the_limits_is_accepted(client, proxy):
    for password in ("x" * 8, "é" * 256):
        assert client.put(PASSWORD_URL, json={"password": password}, headers=as_user("reader")).status_code == 200


def test_the_setting_route_ignores_a_password(client, proxy):
    client.put("/api/settings", json={"opds_password": "reader-password"}, headers=as_user("admin"))
    assert orchestrator.store.get_setting(None, "opds_password_hash:admin") is None


@pytest.mark.parametrize("mode", ["none", "local", "cwa"])
def test_only_the_proxy_mode_has_an_opds_password(client, monkeypatch, mode):
    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, "CWA_URL", "http://localhost:9999")
    monkeypatch.setattr(config, "API_KEY", "key")
    keyed = {"X-Api-Key": "key"}
    response = client.put(PASSWORD_URL, json={"password": "reader-password"}, headers=keyed)
    assert response.status_code == 400
    assert "through the proxy" in response.json["error"]
    info = client.get("/api/settings", headers=keyed).json
    assert (info["opds_user"], info["opds_password_set"]) == (None, False)


def test_a_user_needs_the_login_of_the_proxy_to_set_a_password(client, proxy):
    body = {"password": "reader-password"}
    assert client.put(PASSWORD_URL, json=body, headers=H).status_code == 401
    assert client.put(PASSWORD_URL, json=body, headers={"X-Api-Key": "key"}).status_code == 400
    assert client.put(PASSWORD_URL, json=body, headers={"X-Remote-User": "reader"}).status_code == 403  # no CSRF header
    assert client.put(PASSWORD_URL, json=body, headers=as_user("reader")).status_code == 200
    untrusted = client.put(
        PASSWORD_URL, json=body, headers=as_user("reader"), environ_base={"REMOTE_ADDR": "198.51.100.7"}
    )
    assert untrusted.status_code == 401


def test_config_tells_the_page_the_login_mode(client, monkeypatch):
    assert client.get("/api/config").json["auth_mode"] == "none"
    monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
    monkeypatch.setattr(config, "PROXY_AUTH_TRUSTED_IPS", ["127.0.0.1"])
    assert client.get("/api/config", headers={"X-Remote-User": "reader"}).json["auth_mode"] == "proxy"