Add an OPDS catalog
A reader app can browse the stored books by title, author and last change, search them, and download them. The download and cover routes share their code with the story page. An admin turns the catalog on in the settings. The login follows AUTH_METHOD: none needs no login, local and cwa take the account login, and proxy takes the user name from the proxy with an OPDS password that each user sets. The API key does not open the catalog.
This commit is contained in:
@@ -472,6 +472,9 @@ def test_settings(client, monkeypatch, tmp_path):
|
||||
"pause_uploads": False,
|
||||
"ingest_available": True,
|
||||
"personal_ini": "[defaults]\nis_adult:true\n",
|
||||
"opds_enabled": False,
|
||||
"opds_password_set": False,
|
||||
"opds_user": None,
|
||||
"download_template": "{title_sort} - {authors}",
|
||||
"download_date_format": "%b, %Y",
|
||||
"story_jobs_hours": 36,
|
||||
|
||||
@@ -0,0 +1,267 @@
|
||||
"""OPDS navigation, book lists, downloads and the shared login gate."""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from xml.etree import ElementTree as ET
|
||||
|
||||
import pytest
|
||||
from test_api import client as client
|
||||
from test_api import cwa as cwa
|
||||
from test_build import PNG
|
||||
|
||||
from bindery import diffing, fff, opds, orchestrator
|
||||
|
||||
NS = {"a": opds.ATOM}
|
||||
ACQUIRE = "http://opds-spec.org/acquisition"
|
||||
PATHS = ["/opds", "/opds/", "/opds/title", "/opds/author", "/opds/changed"]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def enabled_catalog(client):
|
||||
orchestrator.store.set_setting(None, "opds_enabled", True)
|
||||
|
||||
|
||||
def feed(response):
|
||||
assert response.status_code == 200
|
||||
assert response.mimetype == "application/atom+xml"
|
||||
assert response.headers["Cache-Control"] == "private, no-store"
|
||||
root = ET.fromstring(response.data)
|
||||
assert root.tag == f"{{{opds.ATOM}}}feed"
|
||||
assert root.findtext("a:id", namespaces=NS)
|
||||
assert root.findtext("a:title", namespaces=NS)
|
||||
assert datetime.fromisoformat(root.findtext("a:updated", namespaces=NS)).tzinfo is not None
|
||||
return root
|
||||
|
||||
|
||||
def links(element):
|
||||
return {link.attrib["rel"]: link.attrib["href"] for link in element.findall("a:link", NS)}
|
||||
|
||||
|
||||
def entries(root):
|
||||
return root.findall("a:entry", NS)
|
||||
|
||||
|
||||
def titles(root):
|
||||
return [entry.findtext("a:title", namespaces=NS) for entry in entries(root)]
|
||||
|
||||
|
||||
def add_book(title="A book", author="Writer", changed=None, url=None, included=True, meta=None):
|
||||
store = orchestrator.store
|
||||
url = url or f"https://example.com/books/{len(store.list_stories()) + 1}"
|
||||
sid = store.create_story(url, "example.com", title, author, [])
|
||||
store.update_story(sid, last_changed=changed, site_values=meta or {})
|
||||
if included:
|
||||
html = "<p>Stored text.</p>"
|
||||
store.add_chapter(sid, url + "/1", "One", html, diffing.text_hash(html), "site", 0)
|
||||
return sid
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def no_site_requests(monkeypatch):
|
||||
def fail(*args, **kwargs):
|
||||
pytest.fail("OPDS must not contact a story site")
|
||||
|
||||
for name in ("fetch_story", "fetch_chapter", "collected_images"):
|
||||
monkeypatch.setattr(fff, name, fail)
|
||||
|
||||
|
||||
def test_navigation_and_empty_lists(client):
|
||||
root = feed(client.get("/opds"))
|
||||
assert titles(root) == [title for title, _, _ in opds.LISTS.values()]
|
||||
assert len({entry.findtext("a:id", namespaces=NS) for entry in entries(root)}) == 3
|
||||
assert client.get("/opds/").data == client.get("/opds").data
|
||||
for entry, (_, kind, text) in zip(entries(root), opds.LISTS.values(), strict=True):
|
||||
assert entry.findtext("a:content", namespaces=NS) == text
|
||||
link = entry.find("a:link", NS)
|
||||
assert link.attrib["type"] == kind
|
||||
books = feed(client.get(link.attrib["href"]))
|
||||
assert entries(books) == []
|
||||
assert links(books)["start"] == links(books)["up"] == "/opds"
|
||||
assert "next" not in links(books)
|
||||
|
||||
|
||||
def test_sort_orders_dates_and_stable_book_ids(client):
|
||||
add_book("The Zebra", "alice", "2024-01-01T12:00:00+00:00")
|
||||
add_book("An Apple", "Bob", "2024-03-01T12:00:00+00:00")
|
||||
add_book("Banana", "ALICE", "2024-02-01T12:00:00+00:00")
|
||||
expected = {
|
||||
"title": ["An Apple", "Banana", "The Zebra"],
|
||||
"author?name=Alice": ["Banana", "The Zebra"],
|
||||
"changed": ["An Apple", "Banana", "The Zebra"],
|
||||
}
|
||||
ids = []
|
||||
for order, names in expected.items():
|
||||
root = feed(client.get(f"/opds/{order}"))
|
||||
assert titles(root) == names
|
||||
if order != "author?name=Alice":
|
||||
assert root.findtext("a:updated", namespaces=NS) == "2024-03-01T12:00:00Z"
|
||||
ids.append({e.findtext("a:id", namespaces=NS) for e in entries(root)})
|
||||
for entry in entries(root):
|
||||
assert entry.findtext("a:author/a:name", namespaces=NS)
|
||||
assert entry.findtext("a:content", namespaces=NS) == "Chapters: 1"
|
||||
assert entry.find("a:link", NS).attrib["type"] == "application/epub+zip"
|
||||
assert ids[0] == ids[2] > ids[1]
|
||||
|
||||
|
||||
def test_dates_use_creation_fallback_and_compare_offsets(client):
|
||||
sid = add_book("Created")
|
||||
created = orchestrator.store.get_story(sid)["created_at"]
|
||||
expected = datetime.fromisoformat(created).astimezone(UTC).isoformat().replace("+00:00", "Z")
|
||||
root = feed(client.get("/opds/changed"))
|
||||
assert entries(root)[0].findtext("a:updated", namespaces=NS) == expected
|
||||
orchestrator.store.update_story(sid, last_changed="2024-01-01T12:00:00+02:00")
|
||||
add_book("Later", changed="2024-01-01T11:00:00+00:00")
|
||||
assert titles(feed(client.get("/opds/changed"))) == ["Later", "Created"]
|
||||
|
||||
|
||||
def test_xml_escaping_unicode_and_missing_metadata(client):
|
||||
add_book('Books <&> "日本語"\x00', "A & B <writers>\x0b")
|
||||
empty = add_book("", "")
|
||||
root = feed(client.get("/opds/title"))
|
||||
assert titles(root)[0] == 'Books <&> "日本語"'
|
||||
assert entries(root)[0].findtext("a:author/a:name", namespaces=NS) == "A & B <writers>"
|
||||
assert titles(root)[1] == orchestrator.store.get_story(empty)["url"]
|
||||
assert entries(root)[1].findtext("a:author/a:name", namespaces=NS) == "Unknown"
|
||||
|
||||
|
||||
def test_only_books_with_included_chapters_appear(client):
|
||||
add_book("Pending", included=False)
|
||||
sid = add_book("Excluded")
|
||||
chapter = orchestrator.store.list_chapters(sid)[0]
|
||||
orchestrator.store.update_chapter(chapter["id"], included=False)
|
||||
add_book("Ready")
|
||||
assert titles(feed(client.get("/opds/title"))) == ["Ready"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("order", ["title", "changed", "author?name=Same", "search?q=same"])
|
||||
def test_pagination_with_ties(client, monkeypatch, order):
|
||||
monkeypatch.setattr(opds, "PAGE_SIZE", 2)
|
||||
for _ in range(5):
|
||||
add_book("Same", "Same", "2024-01-01T00:00:00+00:00")
|
||||
first = feed(client.get(f"/opds/{order}"))
|
||||
second = feed(client.get(links(first)["next"]))
|
||||
last = feed(client.get(links(second)["next"]))
|
||||
assert [len(entries(root)) for root in (first, second, last)] == [2, 2, 1]
|
||||
assert "previous" not in links(first)
|
||||
assert "next" not in links(last)
|
||||
assert links(second)["previous"] == links(first)["self"]
|
||||
assert links(last)["previous"] == links(second)["self"]
|
||||
ids = [e.findtext("a:id", namespaces=NS) for root in (first, second, last) for e in entries(root)]
|
||||
assert ids == [f"https://example.com/books/{i}" for i in range(1, 6)]
|
||||
assert client.get(links(first)["self"].replace("page=1", "page=4")).status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.parametrize("page", ["0", "-1", "1.5", "abc", "", "999999999999999999999"])
|
||||
def test_bad_page(client, page):
|
||||
assert client.get("/opds/title", query_string={"page": page}).status_code == 400
|
||||
|
||||
|
||||
def test_unknown_list(client):
|
||||
assert client.get("/opds/missing").status_code == 404
|
||||
|
||||
|
||||
DC = {"a": opds.ATOM, "dc": opds.DC}
|
||||
|
||||
|
||||
def test_entry_shows_the_book_metadata_and_cover(client):
|
||||
sid = add_book(
|
||||
"Stored title",
|
||||
meta={
|
||||
"title": "The Book",
|
||||
"authors": ["Ann Writer", "Ben Reader"],
|
||||
"comments": "<p>First <b>part</b>.</p><p>Second part.</p>",
|
||||
"tags": ["Space", "Fiction"],
|
||||
"series": "The Saga",
|
||||
"series_index": 2.0,
|
||||
"pubdate": "2020-05-06T12:00:00+00:00",
|
||||
"languages": "en",
|
||||
},
|
||||
)
|
||||
orchestrator.store.update_story(sid, pinned={"title": "The Kept Title"})
|
||||
entry = entries(feed(client.get("/opds/title")))[0]
|
||||
assert entry.findtext("a:title", namespaces=NS) == "The Kept Title"
|
||||
assert [a.findtext("a:name", namespaces=NS) for a in entry.findall("a:author", NS)] == ["Ann Writer", "Ben Reader"]
|
||||
assert entry.find("a:author/a:uri", NS).text == "/opds/author?name=Ann+Writer"
|
||||
assert entry.find("a:content", NS).attrib["type"] == "text"
|
||||
assert entry.findtext("a:content", namespaces=NS) == (
|
||||
"Series: The Saga, book 2\nChapters: 1\n\nFirst part.\n\nSecond part."
|
||||
)
|
||||
assert [c.attrib["term"] for c in entry.findall("a:category", NS)] == ["Space", "Fiction"]
|
||||
assert entry.findtext("dc:language", namespaces=DC) == "en"
|
||||
assert entry.findtext("dc:issued", namespaces=DC) == "2020-05-06"
|
||||
assert entry.findtext("dc:publisher", namespaces=DC) == "example.com"
|
||||
assert "http://opds-spec.org/image" not in links(entry)
|
||||
|
||||
orchestrator.store.put_image(sid, "https://example.com/cover.png", "png", "image/png", PNG, is_cover=True)
|
||||
entry = entries(feed(client.get("/opds/title")))[0]
|
||||
image = links(entry)["http://opds-spec.org/image"]
|
||||
assert image == links(entry)["http://opds-spec.org/image/thumbnail"]
|
||||
assert image.startswith(f"/opds/books/{sid}/cover?v=")
|
||||
response = client.get(image)
|
||||
assert response.data == PNG
|
||||
assert response.mimetype == "image/png"
|
||||
assert "private" in response.headers["Cache-Control"]
|
||||
orchestrator.store.put_image(sid, "https://example.com/new.png", "png", "image/png", PNG, is_cover=True)
|
||||
assert links(entries(feed(client.get("/opds/title")))[0])["http://opds-spec.org/image"] != image
|
||||
assert client.get(f"/opds/books/{add_book()}/cover").status_code == 404
|
||||
|
||||
|
||||
def test_cover_of_an_unsafe_type_is_left_out(client):
|
||||
sid = add_book()
|
||||
orchestrator.store.put_image(sid, "https://example.com/c.svg", "svg", "image/svg+xml", b"<svg/>", is_cover=True)
|
||||
assert "http://opds-spec.org/image" not in links(entries(feed(client.get("/opds/title")))[0])
|
||||
assert client.get(f"/opds/books/{sid}/cover").status_code == 404
|
||||
|
||||
|
||||
def test_authors_in_surname_order_with_their_books(client):
|
||||
add_book("One", "Zed Adams")
|
||||
add_book("Two", "ann writer")
|
||||
add_book("Three", "", meta={"authors": ["Ann Writer", "Bob Brown"]})
|
||||
add_book("Four", "")
|
||||
root = feed(client.get("/opds/author"))
|
||||
assert titles(root) == ["Zed Adams", "Bob Brown", "Unknown", "Ann Writer"]
|
||||
assert [e.findtext("a:content", namespaces=NS) for e in entries(root)] == ["1 book", "1 book", "1 book", "2 books"]
|
||||
assert links(root)["up"] == "/opds"
|
||||
writer = feed(client.get(links(entries(root)[3])["subsection"]))
|
||||
assert titles(writer) == ["Three", "Two"]
|
||||
assert links(writer)["up"] == "/opds/author"
|
||||
assert client.get("/opds/author", query_string={"name": "Nobody"}).status_code == 404
|
||||
|
||||
|
||||
def test_author_list_pages(client, monkeypatch):
|
||||
monkeypatch.setattr(opds, "PAGE_SIZE", 2)
|
||||
for name in ("A One", "B Two", "C Three"):
|
||||
add_book(author=name)
|
||||
first = feed(client.get("/opds/author"))
|
||||
assert titles(first) == ["A One", "C Three"]
|
||||
assert titles(feed(client.get(links(first)["next"]))) == ["B Two"]
|
||||
|
||||
|
||||
def test_search(client):
|
||||
add_book("The Long Road", "Ann Writer", meta={"tags": ["Adventure"]})
|
||||
add_book("Short Road", "Ben Reader", meta={"series": "Roads"})
|
||||
add_book("Other", "Cat Smith")
|
||||
|
||||
def search(q):
|
||||
return titles(feed(client.get("/opds/search", query_string={"q": q})))
|
||||
|
||||
assert search("road") == ["The Long Road", "Short Road"]
|
||||
assert search("ROAD writer") == ["The Long Road"]
|
||||
assert search("adventure") == ["The Long Road"]
|
||||
assert search("roads") == ["Short Road"]
|
||||
assert search("missing") == search("") == search(" ") == []
|
||||
assert feed(client.get("/opds/search")).findtext("a:title", namespaces=NS) == "Search: "
|
||||
|
||||
|
||||
def test_every_feed_links_the_search(client):
|
||||
add_book()
|
||||
for path in [*PATHS, "/opds/author?name=Writer", "/opds/search?q=book"]:
|
||||
searches = [link for link in feed(client.get(path)).findall("a:link", NS) if link.attrib["rel"] == "search"]
|
||||
found = {link.attrib["type"]: link.attrib["href"] for link in searches}
|
||||
assert found == {
|
||||
"application/opensearchdescription+xml": "/opds/search.xml",
|
||||
opds.ACQUISITION: "/opds/search?q={searchTerms}",
|
||||
}
|
||||
response = client.get("/opds/search.xml")
|
||||
assert response.mimetype == "application/opensearchdescription+xml"
|
||||
url = ET.fromstring(response.data).find(f"{{{opds.OPENSEARCH}}}Url")
|
||||
assert url.attrib == {"type": opds.ACQUISITION, "template": "/opds/search?q={searchTerms}"}
|
||||
@@ -0,0 +1,329 @@
|
||||
"""The login of the OPDS catalog: the account login in the local and CWA modes, a password of its own in proxy mode."""
|
||||
|
||||
import base64
|
||||
import io
|
||||
import zipfile
|
||||
|
||||
import pytest
|
||||
from test_api import H, login
|
||||
from test_api import client as client
|
||||
from test_api import cwa as cwa
|
||||
from test_build import URL, metahtml
|
||||
from test_opds import PATHS, add_book, enabled_catalog, entries, feed, links, no_site_requests
|
||||
from werkzeug.security import generate_password_hash
|
||||
|
||||
from bindery import auth, config, opds, orchestrator, web
|
||||
|
||||
__all__ = ["enabled_catalog", "no_site_requests"] # fixtures that every test here uses
|
||||
|
||||
PASSWORD = "reader-password"
|
||||
CHALLENGE = 'Basic realm="Bindery", charset="UTF-8"'
|
||||
ACQUIRE = "http://opds-spec.org/acquisition"
|
||||
GATED = [*PATHS, "/opds/search?q=a", "/opds/search.xml", "/opds/books/1.epub", "/opds/books/1/cover"]
|
||||
OTHER_LOGINS = [
|
||||
{"Authorization": "Bearer api-secret"},
|
||||
{"X-Api-Key": "api-secret"},
|
||||
{"X-Remote-User": "admin"},
|
||||
]
|
||||
|
||||
|
||||
def basic(password, user="reader"):
|
||||
return {"Authorization": "Basic " + base64.b64encode(f"{user}:{password}".encode()).decode()}
|
||||
|
||||
|
||||
def set_password(password=PASSWORD, user="reader"):
|
||||
hashed = generate_password_hash(password, method="pbkdf2:sha256:1000")
|
||||
orchestrator.store.set_setting(None, orchestrator._opds_password_name(user), hashed)
|
||||
|
||||
|
||||
def fresh_lockout(monkeypatch):
|
||||
monkeypatch.setattr(auth, "limiter", auth.LoginLimiter())
|
||||
|
||||
|
||||
@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, "API_KEY", "api-secret")
|
||||
set_password()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def local(monkeypatch):
|
||||
monkeypatch.setattr(config, "AUTH_METHOD", "local")
|
||||
monkeypatch.setattr(config, "LOCAL_USERNAME", "reader")
|
||||
monkeypatch.setattr(config, "LOCAL_PASSWORD_HASH", generate_password_hash("web-pw", method="pbkdf2:sha256:1000"))
|
||||
monkeypatch.setattr(config, "API_KEY", "api-secret")
|
||||
set_password() # an OPDS password does not apply here
|
||||
|
||||
|
||||
def test_proxy_mode_takes_the_proxy_user_name_with_the_password_of_that_user(client, proxy, monkeypatch):
|
||||
add_book()
|
||||
set_password("alice-password", "alice")
|
||||
for path in GATED:
|
||||
response = client.get(path)
|
||||
assert response.status_code == 401
|
||||
assert response.headers["WWW-Authenticate"] == CHALLENGE
|
||||
for path in PATHS:
|
||||
response = client.get(path, headers=basic(PASSWORD, "reader"))
|
||||
feed(response)
|
||||
assert "Set-Cookie" not in response.headers
|
||||
assert client.get("/opds/search.xml", headers=basic("alice-password", "alice")).status_code == 200
|
||||
# The user name matches without regard to case and spaces. Each user has their own password.
|
||||
assert client.get("/opds", headers=basic(PASSWORD, " READER ")).status_code == 200
|
||||
for headers in [
|
||||
basic(PASSWORD, "alice"),
|
||||
basic("alice-password", "reader"),
|
||||
basic(PASSWORD, "nobody"),
|
||||
basic(PASSWORD, ""),
|
||||
*OTHER_LOGINS,
|
||||
basic("api-secret"),
|
||||
{**OTHER_LOGINS[2], **basic("wrong")},
|
||||
{"X-Remote-User": "alice", **basic(PASSWORD, "nobody")},
|
||||
]:
|
||||
fresh_lockout(monkeypatch)
|
||||
for path in PATHS[:3]:
|
||||
assert client.get(path, headers=headers).status_code == 401
|
||||
|
||||
|
||||
def test_the_catalog_password_does_not_open_the_api(client, proxy):
|
||||
for headers in (basic(PASSWORD), {"Authorization": f"Bearer {PASSWORD}"}, {"X-Api-Key": PASSWORD}):
|
||||
response = client.get("/api/stories", headers=headers)
|
||||
assert response.status_code == 401
|
||||
assert "WWW-Authenticate" not in response.headers
|
||||
assert client.get("/api/stories", headers={"X-Api-Key": "api-secret"}).status_code == 200
|
||||
assert client.post("/api/stories", json={"urls": []}, headers={**H, **basic(PASSWORD)}).status_code == 401
|
||||
|
||||
|
||||
def test_a_user_without_a_password_meets_the_same_work_as_one_with_a_password(client, proxy, monkeypatch):
|
||||
calls = []
|
||||
real = auth.check_password_hash
|
||||
monkeypatch.setattr(auth, "check_password_hash", lambda *args: calls.append(args[0]) or real(*args))
|
||||
for user in ("nobody", "reader"):
|
||||
calls.clear()
|
||||
response = client.get("/opds", headers=basic("wrong-password", user))
|
||||
assert response.status_code == 401
|
||||
assert response.json == {"error": "Unauthorized"}
|
||||
assert len(calls) == 1
|
||||
|
||||
|
||||
def test_local_mode_takes_the_account_login(client, local, monkeypatch):
|
||||
for path in GATED:
|
||||
response = client.get(path)
|
||||
assert response.status_code == 401
|
||||
assert response.headers["WWW-Authenticate"] == CHALLENGE
|
||||
assert response.json["error"] == "Unauthorized"
|
||||
for path in PATHS:
|
||||
response = client.get(path, headers=basic("web-pw"))
|
||||
feed(response)
|
||||
assert "Set-Cookie" not in response.headers
|
||||
for headers in [basic(PASSWORD), basic("web-pw", "other"), basic("wrong"), *OTHER_LOGINS]:
|
||||
fresh_lockout(monkeypatch)
|
||||
assert client.get("/opds", headers=headers).status_code == 401
|
||||
assert client.get("/opds", headers=basic("web-pw", " reader ")).status_code == 200
|
||||
|
||||
|
||||
def test_local_mode_takes_the_web_session_but_not_the_api_key_with_it(client, local):
|
||||
assert client.get("/opds").status_code == 401
|
||||
assert login(client, password="web-pw").status_code == 200
|
||||
for path in PATHS:
|
||||
feed(client.get(path))
|
||||
assert client.get("/opds", headers={"X-Api-Key": "api-secret"}).status_code == 200 # the session counts
|
||||
client.post("/api/auth/logout", headers=H)
|
||||
assert client.get("/opds").status_code == 401
|
||||
assert client.get("/opds", headers={"X-Api-Key": "api-secret"}).status_code == 401
|
||||
|
||||
|
||||
def test_cwa_mode_takes_the_cwa_login(client, cwa, monkeypatch):
|
||||
monkeypatch.setattr(config, "API_KEY", "api-secret")
|
||||
set_password()
|
||||
assert client.get("/opds").status_code == 401
|
||||
assert client.get("/opds", headers=basic("pw")).status_code == 200
|
||||
for headers in [basic("wrong"), basic("pw", "other"), basic(PASSWORD), *OTHER_LOGINS]:
|
||||
fresh_lockout(monkeypatch)
|
||||
assert client.get("/opds", headers=headers).status_code == 401
|
||||
assert login(client).status_code == 200
|
||||
feed(client.get("/opds/title"))
|
||||
client.post("/api/auth/logout", headers=H)
|
||||
monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["someone"])
|
||||
assert client.get("/opds", headers=basic("pw")).status_code == 401
|
||||
|
||||
|
||||
def test_none_mode_needs_no_login(client):
|
||||
assert orchestrator.store.get_setting(None, "opds_password_hash") is None
|
||||
for path in PATHS:
|
||||
feed(client.get(path))
|
||||
set_password()
|
||||
feed(client.get("/opds"))
|
||||
|
||||
|
||||
def test_a_good_account_login_is_kept_for_five_minutes(client, local, monkeypatch):
|
||||
clock = [1000.0]
|
||||
monkeypatch.setattr(auth.time, "monotonic", lambda: clock[0])
|
||||
calls = []
|
||||
real = auth.login_local
|
||||
monkeypatch.setattr(auth, "login_local", lambda *args: calls.append(args) or real(*args))
|
||||
for seconds in (0, auth.KEPT_SECONDS - 1, 1):
|
||||
clock[0] += seconds
|
||||
assert client.get("/opds", headers=basic("web-pw")).status_code == 200
|
||||
assert len(calls) == 2
|
||||
# A wrong password is never kept. A new web password ends a kept login.
|
||||
assert client.get("/opds", headers=basic("wrong")).status_code == 401
|
||||
assert client.get("/opds", headers=basic("wrong")).status_code == 401
|
||||
assert len(calls) == 4
|
||||
monkeypatch.setattr(config, "LOCAL_PASSWORD_HASH", generate_password_hash("new-pw", method="pbkdf2:sha256:1000"))
|
||||
assert client.get("/opds", headers=basic("web-pw")).status_code == 401
|
||||
assert client.get("/opds", headers=basic("new-pw")).status_code == 200
|
||||
|
||||
|
||||
def test_a_good_catalog_password_is_kept_and_a_new_password_ends_it(client, proxy, monkeypatch):
|
||||
calls = []
|
||||
real = auth.check_password_hash
|
||||
monkeypatch.setattr(auth, "check_password_hash", lambda *args: calls.append(args) or real(*args))
|
||||
for _ in range(3):
|
||||
assert client.get("/opds", headers=basic(PASSWORD)).status_code == 200
|
||||
assert len(calls) == 1
|
||||
set_password("a new password")
|
||||
assert client.get("/opds", headers=basic(PASSWORD)).status_code == 401
|
||||
assert client.get("/opds", headers=basic("a new password")).status_code == 200
|
||||
|
||||
|
||||
def test_a_disabled_catalog_answers_404_before_any_login(client, proxy):
|
||||
orchestrator.store.set_setting(None, "opds_enabled", False)
|
||||
for headers in ({}, basic(PASSWORD), basic("wrong")):
|
||||
assert client.get("/opds", headers=headers).status_code == 404
|
||||
|
||||
|
||||
def test_wrong_catalog_passwords_lock_out_a_user_at_an_address(client, proxy):
|
||||
set_password("alice-password", "alice")
|
||||
for _ in range(auth.MAX_FAILURES):
|
||||
assert client.get("/opds", headers=basic("wrong")).status_code == 401
|
||||
assert client.get("/opds", headers=basic(PASSWORD)).status_code == 401
|
||||
# Another user at that address, and the same user at another address, can log in.
|
||||
assert client.get("/opds", headers=basic("alice-password", "alice")).status_code == 200
|
||||
other = {"environ_base": {"REMOTE_ADDR": "198.51.100.7"}}
|
||||
assert client.get("/opds", headers=basic(PASSWORD), **other).status_code == 200
|
||||
|
||||
|
||||
def test_wrong_account_passwords_lock_out_an_address(client, local):
|
||||
for _ in range(auth.MAX_FAILURES):
|
||||
assert client.get("/opds", headers=basic("wrong")).status_code == 401
|
||||
assert client.get("/opds", headers=basic("web-pw")).status_code == 401
|
||||
assert client.get("/opds", headers=basic("web-pw"), environ_base={"REMOTE_ADDR": "198.51.100.7"}).status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
["Basic", "Basic !!!", "Digest cmVhZGVyOnB3", basic("")["Authorization"], basic("x" * 257)["Authorization"]]
|
||||
+ ["Basic " + base64.b64encode(text).decode() for text in (b"reader", b"\xff:pw")],
|
||||
)
|
||||
def test_bad_basic_headers(client, proxy, monkeypatch, value):
|
||||
def fail(*args):
|
||||
pytest.fail("A bad header must not reach the hash")
|
||||
|
||||
monkeypatch.setattr(auth, "check_password_hash", fail)
|
||||
assert client.get("/opds", headers={"Authorization": value}).status_code == 401
|
||||
|
||||
|
||||
def test_bad_config_fails_closed(client, monkeypatch):
|
||||
set_password()
|
||||
monkeypatch.setattr(config, "AUTH_METHOD", "invalid")
|
||||
for path in PATHS:
|
||||
assert client.get(path, headers=basic(PASSWORD)).status_code == 503
|
||||
|
||||
|
||||
def test_host_protection(client):
|
||||
assert client.get("/opds", base_url="http://attacker.example").status_code == 400
|
||||
|
||||
|
||||
def test_url_base_and_download_auth(client, proxy, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(web.app, "wsgi_app", web.PrefixMiddleware(web.app.wsgi_app, "/bindery"))
|
||||
monkeypatch.setattr(opds, "PAGE_SIZE", 1)
|
||||
sid = add_book()
|
||||
add_book("Second")
|
||||
headers = basic(PASSWORD)
|
||||
root = feed(client.get("/bindery/opds", headers=headers))
|
||||
for entry in entries(root):
|
||||
books = feed(client.get(links(entry)["subsection"], headers=headers))
|
||||
assert all(href.startswith("/bindery/") for href in links(books).values())
|
||||
if entry.find("a:link", {"a": opds.ATOM}).attrib["type"] == opds.ACQUISITION:
|
||||
assert client.get(links(books)["next"], headers=headers).status_code == 200
|
||||
books = feed(client.get("/bindery/opds/title", headers=headers))
|
||||
download = links(entries(books)[0])[ACQUIRE]
|
||||
assert download == f"/bindery/opds/books/{sid}.epub"
|
||||
assert client.get(download).status_code == 401
|
||||
assert client.get(download, headers=basic("wrong")).status_code == 401
|
||||
file = tmp_path / "book.epub"
|
||||
file.write_bytes(b"epub bytes")
|
||||
monkeypatch.setattr(orchestrator, "build_for_download", lambda story_id: (file.open("rb"), "book.epub"))
|
||||
with client.get(download, headers=headers) as response:
|
||||
assert response.data == b"epub bytes"
|
||||
assert response.mimetype == "application/epub+zip"
|
||||
assert client.get("/opds", headers=headers).status_code == 404
|
||||
|
||||
|
||||
def test_download_builds_from_store_without_changing_chapters(client, local, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(config, "TMP_DIR", tmp_path)
|
||||
sid = add_book(url=URL)
|
||||
orchestrator.store.update_story(sid, metahtml=metahtml())
|
||||
before = orchestrator.store.chapters_for_build(sid)
|
||||
root = feed(client.get("/opds/title", headers=basic("web-pw")))
|
||||
with client.get(links(entries(root)[0])[ACQUIRE], headers=basic("web-pw")) as response:
|
||||
assert response.status_code == 200
|
||||
assert response.mimetype == "application/epub+zip"
|
||||
with zipfile.ZipFile(io.BytesIO(response.data)) as epub:
|
||||
assert epub.read("mimetype") == b"application/epub+zip"
|
||||
assert orchestrator.store.chapters_for_build(sid) == before
|
||||
|
||||
|
||||
def test_the_catalog_takes_no_changes(client, proxy):
|
||||
assert client.post("/opds", headers=basic(PASSWORD)).status_code == 405
|
||||
|
||||
|
||||
def test_an_opds_download_is_named_by_the_download_template(client, local, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(config, "TMP_DIR", tmp_path)
|
||||
sid = add_book(url=URL, meta={"title": "The Book", "authors": ["Ann Writer"]})
|
||||
orchestrator.store.update_story(sid, metahtml=metahtml())
|
||||
root = feed(client.get("/opds/title", headers=basic("web-pw")))
|
||||
href = links(entries(root)[0])[ACQUIRE]
|
||||
|
||||
def disposition():
|
||||
with client.get(href, headers=basic("web-pw")) as response:
|
||||
assert response.status_code == 200
|
||||
return response.headers["Content-Disposition"]
|
||||
|
||||
assert 'filename="Book, The - Ann Writer.epub"' in disposition()
|
||||
orchestrator.store.set_setting(None, "download_template", "{title} - {authors}")
|
||||
assert 'filename="The Book - Ann Writer.epub"' in disposition()
|
||||
|
||||
|
||||
def test_forget_clears_the_wrong_passwords_of_one_user_from_every_address():
|
||||
limiter = auth.LoginLimiter()
|
||||
for address in ("10.0.0.1", "10.0.0.2"):
|
||||
for _ in range(auth.MAX_FAILURES):
|
||||
limiter.failed("Alice", address)
|
||||
limiter.failed("bob", address)
|
||||
limiter.forget("ALICE")
|
||||
assert not limiter.locked("alice", "10.0.0.1")
|
||||
assert not limiter.locked("alice", "10.0.0.2")
|
||||
assert limiter.locked("bob", "10.0.0.1")
|
||||
|
||||
|
||||
def test_setting_a_password_ends_the_lockout_of_that_user(client, proxy):
|
||||
set_password("bob-password", "bob")
|
||||
far = {"environ_base": {"REMOTE_ADDR": "198.51.100.7"}}
|
||||
# A user without a password tries from two addresses, and meets the lockout.
|
||||
for kwargs in ({}, far):
|
||||
for _ in range(auth.MAX_FAILURES):
|
||||
assert client.get("/opds", headers=basic("guess", "alice"), **kwargs).status_code == 401
|
||||
for _ in range(auth.MAX_FAILURES):
|
||||
assert client.get("/opds", headers=basic("guess", "bob")).status_code == 401
|
||||
assert client.get("/opds", headers=basic("alice-password", "alice")).status_code == 401
|
||||
saved = client.put(
|
||||
"/api/settings/opds-password", json={"password": "alice-password"}, headers={**H, "X-Remote-User": "Alice"}
|
||||
)
|
||||
assert saved.status_code == 200
|
||||
for kwargs in ({}, far):
|
||||
assert client.get("/opds", headers=basic("alice-password", "alice"), **kwargs).status_code == 200
|
||||
# The lockout of another user stays.
|
||||
assert client.get("/opds", headers=basic("bob-password", "bob")).status_code == 401
|
||||
@@ -0,0 +1,186 @@
|
||||
"""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"
|
||||
Reference in New Issue
Block a user