From 28cd0dcde93613d60f25295143a4da35974c8c54 Mon Sep 17 00:00:00 2001 From: Timothy Allen Date: Sat, 3 Oct 2026 11:13:51 +0200 Subject: [PATCH] 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. --- bindery/auth.py | 134 +++++++++++++-- bindery/opds.py | 297 ++++++++++++++++++++++++++++++++ bindery/orchestrator.py | 41 ++++- bindery/store.py | 12 ++ bindery/web.py | 60 ++++--- tests/test_api.py | 3 + tests/test_opds.py | 267 +++++++++++++++++++++++++++++ tests/test_opds_auth.py | 329 ++++++++++++++++++++++++++++++++++++ tests/test_opds_settings.py | 186 ++++++++++++++++++++ 9 files changed, 1295 insertions(+), 34 deletions(-) create mode 100644 bindery/opds.py create mode 100644 tests/test_opds.py create mode 100644 tests/test_opds_auth.py create mode 100644 tests/test_opds_settings.py diff --git a/bindery/auth.py b/bindery/auth.py index 2dfdfda..860a4b9 100644 --- a/bindery/auth.py +++ b/bindery/auth.py @@ -1,23 +1,26 @@ -"""Logins: none, one local account, CWA accounts, or a proxy. API keys work in every mode. +"""Logins: none, one local account, CWA accounts, or a proxy. API keys work in every mode. The OPDS catalog takes +the account login, and in proxy mode a password of its own. CWA accounts use an optional read-only app.db, with OPDS as the fallback when the database cannot answer. A rejected database login never falls back. The OPDS login works only with a CWA that refuses a wrong password. """ +import base64 import hashlib import hmac import ipaddress import logging +import secrets import threading import time -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import dataclass from functools import lru_cache from typing import Any from urllib.parse import urlsplit -from werkzeug.security import check_password_hash +from werkzeug.security import check_password_hash, generate_password_hash from bindery import calibre_db, config from bindery.cwa import CwaClient, CwaError @@ -163,6 +166,20 @@ def _api_key_identity(headers: Mapping[str, str]) -> Identity | None: return None +def _session_identity(session: Mapping[str, Any]) -> Identity | None: + """The user of a web session in the local or CWA mode.""" + mode = config.AUTH_METHOD + name = session.get("user") + if not name or session.get("mode") != mode: + return None + if mode == AUTH_LOCAL: + if name == config.LOCAL_USERNAME and session.get("local_password_version") == local_password_version(): + return Identity(name, "session") + elif mode == AUTH_CWA and _allowed(name): + return Identity(name, "session") + return None + + def current_identity( headers: Mapping[str, str], session: Mapping[str, Any], remote_addr: str | None = None ) -> Identity | None: @@ -181,15 +198,7 @@ def current_identity( logger.warning(f"Ignored the user header from {remote_addr}, which is not in PROXY_AUTH_TRUSTED_IPS") return None return Identity(name, "proxy") if name else None - name = session.get("user") - if not name or session.get("mode") != mode: - return None - if mode == AUTH_LOCAL: - if name == config.LOCAL_USERNAME and session.get("local_password_version") == local_password_version(): - return Identity(name, "session") - elif mode == AUTH_CWA and _allowed(name): - return Identity(name, "session") - return None + return _session_identity(session) def client_address(headers: Mapping[str, str], remote_addr: str | None) -> str: @@ -233,6 +242,11 @@ class LoginLimiter: kept = [x for x in self._failures.get(address, []) if x[1] != name.casefold()] self._failures[address] = kept + def forget(self, name: str) -> None: + """Clear the wrong passwords of a user name from every address.""" + with self._lock: + self._failures = {a: [x for x in f if x[1] != name.casefold()] for a, f in self._failures.items()} + limiter = LoginLimiter() @@ -291,3 +305,99 @@ def login_cwa(username: str, password: str, client: CwaClient, address: str) -> raise AuthError("Wrong user name or password") limiter.clear(username, address) return Identity(username, "session") + + +KEPT_SECONDS = 5 * 60 +KEPT_LIMIT = 100 +MAX_PASSWORD = 256 +_kept_key = secrets.token_bytes(32) +_kept_logins: dict[bytes, float] = {} # keyed hash of a good login: end time +_kept_lock = threading.Lock() + + +def _basic_login(headers: Mapping[str, str]) -> tuple[str, str] | None: + """The user name and password of an HTTP Basic login, or None.""" + scheme, _, value = headers.get("Authorization", "").partition(" ") + if scheme.lower() != "basic": + return None + try: + username, colon, password = base64.b64decode(value.strip(), validate=True).decode().partition(":") + except ValueError: + return None + return (username, password) if colon and password and len(password) <= MAX_PASSWORD else None + + +def _kept(parts: tuple[Any, ...], check: Callable[[], bool]) -> bool: + """Whether `check` accepts the login. Bindery keeps a good login for five minutes, so a reader that loads 50 + covers checks a password hash once. `parts` holds everything that the check depends on.""" + key = hmac.digest(_kept_key, repr(parts).encode(), "sha256") + with _kept_lock: + if _kept_logins.get(key, 0) > time.monotonic(): + return True + if not check(): + return False + with _kept_lock: + if len(_kept_logins) >= KEPT_LIMIT: + _kept_logins.clear() + _kept_logins[key] = time.monotonic() + KEPT_SECONDS + return True + + +def _account_login(username: str, password: str, address: str, client: Callable[[], CwaClient]) -> bool: + """Whether this is a good login of the local or the CWA account. A wrong password counts toward the lockout.""" + try: + if config.AUTH_METHOD == AUTH_LOCAL: + login_local(username, password, address) + else: + login_cwa(username, password, client(), address) + except AuthError: + return False + return True + + +@lru_cache(maxsize=1) +def _dummy_hash() -> str: + return generate_password_hash("not a password") + + +def _own_password_login(username: str, password: str, address: str, password_hash: str | None) -> bool: + """Whether this is the OPDS password of this user. A wrong password counts toward the lockout.""" + if limiter.locked(username, address): + return False + # A user without a password meets the same work as one with a password, so the time shows no difference. + if not (check_password_hash(password_hash or _dummy_hash(), password) and password_hash): + limiter.failed(username, address) + return False + limiter.clear(username, address) + return True + + +def catalog_login( + headers: Mapping[str, str], + session: Mapping[str, Any], + address: str, + password_hash_of: Callable[[str], str | None], + client: Callable[[], CwaClient], +) -> bool: + """Whether the request may open the OPDS catalog. + + The local and CWA modes take the web session or the account login in HTTP Basic. The proxy mode takes the user + name that the proxy gives for the web login, with the OPDS password of that user, because the proxy skips its + login on these paths and a client can send any user header there. The API key opens the catalog in no mode. + """ + mode = config.AUTH_METHOD + if mode == AUTH_NONE: + return True + if mode in (AUTH_LOCAL, AUTH_CWA) and _session_identity(session): + return True + login = _basic_login(headers) + if not login: + return False + username, password = login + if mode == AUTH_PROXY: + username = username.strip() + password_hash = password_hash_of(username) + parts = (mode, username.casefold(), password_hash, password) + return _kept(parts, lambda: _own_password_login(username, password, address, password_hash)) + parts = (mode, username, password, local_password_version(), tuple(config.CWA_ALLOWED_USERS)) + return _kept(parts, lambda: _account_login(username, password, address, client)) diff --git a/bindery/opds.py b/bindery/opds.py new file mode 100644 index 0000000..e435f75 --- /dev/null +++ b/bindery/opds.py @@ -0,0 +1,297 @@ +"""An OPDS 1.2 catalog of the books stored in Bindery.""" + +import hashlib +import io +import os +import re +from collections.abc import Callable +from datetime import UTC, datetime +from typing import Any +from urllib.parse import quote +from xml.etree import ElementTree as ET + +from flask import Blueprint, Response, abort, request, send_file, url_for + +from bindery import diffing, filename_template, orchestrator +from bindery.store import sort_title + +bp = Blueprint("opds", __name__) +ATOM = "http://www.w3.org/2005/Atom" +DC = "http://purl.org/dc/terms/" +OPENSEARCH = "http://a9.com/-/spec/opensearch/1.1/" +NAVIGATION = "application/atom+xml;profile=opds-catalog;kind=navigation" +ACQUISITION = "application/atom+xml;profile=opds-catalog;kind=acquisition" +EPUB = "application/epub+zip" +PAGE_SIZE = 50 +UNKNOWN_AUTHOR = "Unknown" # as in Calibre +# Each list: its title, its kind, and the text of its entry in the start feed. +LISTS = { + "title": ("By Title", ACQUISITION, "All books in title order"), + "author": ("By Author", NAVIGATION, "The authors, with the books of each"), + "changed": ("By Last Changed", ACQUISITION, "All books, with the most recently changed books first"), +} +# XML 1.0 excludes control characters, even in escaped text. +INVALID_XML = re.compile("[^\x09\x0a\x0d\x20-퟿-�\U00010000-\U0010ffff]") + + +def _element(parent: ET.Element, tag: str, text: str = "", **attributes: str) -> ET.Element: + element = ET.SubElement(parent, tag, attributes) + element.text = INVALID_XML.sub("", text) + return element + + +def _updated(book: dict[str, Any]) -> str: + # Store timestamps without an offset use the server's local time. + date = datetime.fromisoformat(book["last_changed"] or book["created_at"]) + return date.astimezone(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") + + +def _link(parent: ET.Element, rel: str, href: str, media_type: str) -> None: + _element(parent, "link", rel=rel, href=href, type=media_type) + + +def _feed(title: str, identifier: str, updated: str, href: str, media_type: str) -> ET.Element: + feed = ET.Element("feed", {"xmlns": ATOM, "xmlns:dc": DC}) + _element(feed, "id", identifier) + _element(feed, "title", title) + _element(feed, "updated", updated) + _element(_element(feed, "author"), "name", "Bindery") + _link(feed, "self", href, media_type) + _link(feed, "start", url_for("opds.catalog"), NAVIGATION) + _link(feed, "search", url_for("opds.search_description"), "application/opensearchdescription+xml") + # Many readers take the search template from the feed and never read the description. + _link(feed, "search", f"{url_for('opds.search')}?q={{searchTerms}}", ACQUISITION) + return feed + + +def _response(root: ET.Element, media_type: str) -> Response: + response = Response(ET.tostring(root, encoding="utf-8", xml_declaration=True), content_type=media_type) + response.headers["Cache-Control"] = "private, no-store" + return response + + +def _languages(value: Any) -> list[str]: + return [value] if isinstance(value, str) else list(value or []) + + +def _books() -> list[dict[str, Any]]: + """The books of the catalog, in title order.""" + books = [] + for story in orchestrator.store.catalog(): + meta = orchestrator.book_metadata(story) + title = (meta["title"] or "").strip() or story["url"] + authors = [a.strip() for a in meta["authors"] if a and a.strip()] or [UNKNOWN_AUTHOR] + cover = story["cover_key"] if story["cover_mime"] in orchestrator.COVER_TYPES else None + books.append( + { + **story, + "meta": meta, + "title": title, + "authors": authors, + "updated": _updated(story), + "cover": hashlib.sha256(cover.encode()).hexdigest()[:16] if cover else None, + } + ) + # The ID breaks ties, so a book keeps its page. + return sorted(books, key=lambda book: (sort_title(book["title"]), book["id"])) + + +def _latest(books: list[dict[str, Any]]) -> str: + return max((book["updated"] for book in books), default="1970-01-01T00:00:00Z") + + +def _content(book: dict[str, Any]) -> str: + meta = book["meta"] + lines = [] + if meta.get("series"): + index = filename_template.series_index(meta.get("series_index")) + lines.append(f"Series: {meta['series']}" + (f", book {index}" if index else "")) + lines.append(f"Chapters: {book['chapter_count']}") + description = "\n\n".join(diffing.html_to_blocks(meta.get("comments") or "")) + return "\n".join(lines) + (f"\n\n{description}" if description else "") + + +def _book_entry(feed: ET.Element, book: dict[str, Any]) -> None: + meta = book["meta"] + entry = _element(feed, "entry") + _element(entry, "id", book["url"]) + _element(entry, "title", book["title"]) + _element(entry, "updated", book["updated"]) + for name in book["authors"]: + author = _element(entry, "author") + _element(author, "name", name) + _element(author, "uri", url_for("opds.books", order="author", name=name)) + _element(entry, "dc:publisher", book["site"]) + if meta.get("pubdate"): + _element(entry, "dc:issued", meta["pubdate"][:10]) + for language in _languages(meta.get("languages")): + _element(entry, "dc:language", language) + for tag in meta.get("tags") or []: + _element(entry, "category", term=tag, label=tag) + _element(entry, "content", _content(book), type="text") + _link(entry, "http://opds-spec.org/acquisition", url_for("opds.download", story_id=book["id"]), EPUB) + if book["cover"]: + href = url_for("opds.cover", story_id=book["id"], v=book["cover"]) + for rel in ("http://opds-spec.org/image", "http://opds-spec.org/image/thumbnail"): + _link(entry, rel, href, book["cover_mime"]) + + +def _author_entry(feed: ET.Element, author: dict[str, Any]) -> None: + entry = _element(feed, "entry") + _element(entry, "id", f"urn:bindery:author:{quote(author['name'])}") + _element(entry, "title", author["name"]) + _element(entry, "updated", _latest(author["books"])) + count = len(author["books"]) + _element(entry, "content", f"{count} book" + ("" if count == 1 else "s"), type="text") + _link(entry, "subsection", url_for("opds.books", order="author", name=author["name"]), ACQUISITION) + + +def _authors(books: list[dict[str, Any]]) -> list[dict[str, Any]]: + """The authors of the books, in Calibre's author sort order. A name in other capitals is the same author.""" + authors: dict[str, dict[str, Any]] = {} + for book in books: + for name in book["authors"]: + authors.setdefault(name.casefold(), {"name": name, "books": []})["books"].append(book) + return sorted(authors.values(), key=lambda a: (filename_template.author_sort(a["name"]).casefold(), a["name"])) + + +def _page() -> int: + text = request.args.get("page", "1") + if not re.fullmatch(r"[0-9]{1,9}", text) or int(text) < 1: + raise ValueError("The page must be a positive whole number") + return int(text) + + +def _paged( + title: str, + identifier: str, + items: list[dict[str, Any]], + kind: str, + entry: Callable[[ET.Element, dict[str, Any]], None], + up: str, + endpoint: str, + **args: str, +) -> Response: + """A feed that shows one page of the items. `entry` adds the entry of one item.""" + page = _page() + pages = max(1, (len(items) + PAGE_SIZE - 1) // PAGE_SIZE) + if page > pages: + abort(404) + books = [book for item in items for book in item.get("books", [item])] + feed = _feed(title, identifier, _latest(books), url_for(endpoint, **args, page=page), kind) + _link(feed, "up", up, NAVIGATION) + if page > 1: + _link(feed, "previous", url_for(endpoint, **args, page=page - 1), kind) + if page < pages: + _link(feed, "next", url_for(endpoint, **args, page=page + 1), kind) + for item in items[(page - 1) * PAGE_SIZE : page * PAGE_SIZE]: + entry(feed, item) + return _response(feed, kind) + + +@bp.get("/opds/") +@bp.get("/opds") +def catalog() -> Response: + updated = _latest(_books()) + feed = _feed("Bindery", "urn:bindery:catalog", updated, url_for("opds.catalog"), NAVIGATION) + for order, (title, kind, text) in LISTS.items(): + entry = _element(feed, "entry") + _element(entry, "id", f"urn:bindery:catalog:{order}") + _element(entry, "title", title) + _element(entry, "updated", updated) + _element(entry, "content", text, type="text") + _link(entry, "subsection", url_for("opds.books", order=order), kind) + return _response(feed, NAVIGATION) + + +@bp.get("/opds/") +def books(order: str) -> Response: + if order not in LISTS: + abort(404) + books = _books() + up = url_for("opds.catalog") + if order == "author": + authors = _authors(books) + name = request.args.get("name") + if name is None: + title, kind, _ = LISTS[order] + return _paged( + title, "urn:bindery:catalog:author", authors, kind, _author_entry, up, "opds.books", order=order + ) + author = next((a for a in authors if a["name"].casefold() == name.casefold()), None) + if author is None: + abort(404) + identifier = f"urn:bindery:author:{quote(author['name'])}" + up = url_for("opds.books", order="author") + return _paged( + author["name"], + identifier, + author["books"], + ACQUISITION, + _book_entry, + up, + "opds.books", + order=order, + name=author["name"], + ) + if order == "changed": + books.sort(key=lambda book: book["updated"], reverse=True) + identifier = f"urn:bindery:catalog:{order}" + return _paged(LISTS[order][0], identifier, books, ACQUISITION, _book_entry, up, "opds.books", order=order) + + +@bp.get("/opds/search") +def search() -> Response: + """The books whose title, authors, series or tags hold each word of the search, in title order.""" + query = request.args.get("q", "") + words = query.casefold().split() + found = [] + for book in _books() if words else []: + meta = book["meta"] + text = " ".join([book["title"], *book["authors"], meta.get("series") or "", *(meta.get("tags") or [])]) + if all(word in text.casefold() for word in words): + found.append(book) + identifier = f"urn:bindery:search:{quote(query)}" + up = url_for("opds.catalog") + return _paged(f"Search: {query}", identifier, found, ACQUISITION, _book_entry, up, "opds.search", q=query) + + +@bp.get("/opds/search.xml") +def search_description() -> Response: + root = ET.Element("OpenSearchDescription", xmlns=OPENSEARCH) + _element(root, "ShortName", "Bindery") + _element(root, "Description", "Search the books in Bindery by title, author, series or tag") + _element(root, "InputEncoding", "UTF-8") + _element(root, "OutputEncoding", "UTF-8") + _element(root, "Url", type=ACQUISITION, template=f"{url_for('opds.search')}?q={{searchTerms}}") + return _response(root, "application/opensearchdescription+xml") + + +def send_epub(story_id: int) -> Response: + """The ePub of a story as a download, built from the stored chapters.""" + file, name = orchestrator.build_for_download(story_id) + response = send_file(file, download_name=name, as_attachment=True, mimetype=EPUB) + response.content_length = os.fstat(file.fileno()).st_size + return response + + +def send_cover(story_id: int) -> Response: + cover = orchestrator.story_cover(story_id) + if not cover: + raise KeyError("This story has no cover") + # The URL holds the cover version, so a client can keep the image for a year. + response = send_file(io.BytesIO(cover["data"]), mimetype=cover["mime"], max_age=365 * 24 * 60 * 60) + # A cover needs a login, so a shared cache must not keep it. + response.cache_control.public = False + response.cache_control.private = True + return response + + +@bp.get("/opds/books/.epub") +def download(story_id: int) -> Response: + return send_epub(story_id) + + +@bp.get("/opds/books//cover") +def cover(story_id: int) -> Response: + return send_cover(story_id) diff --git a/bindery/orchestrator.py b/bindery/orchestrator.py index 11198bb..7f99310 100644 --- a/bindery/orchestrator.py +++ b/bindery/orchestrator.py @@ -11,6 +11,8 @@ from datetime import datetime, timedelta from pathlib import Path from typing import Any, BinaryIO +from werkzeug.security import generate_password_hash + from bindery import config, fff, filename_template from bindery.backup import daily_backup from bindery.cwa import Book, BookFinder, CwaClient, CwaError @@ -379,6 +381,7 @@ def get_settings() -> dict[str, Any]: **_upload_state(), "linking_available": _linking_available(), "hold_new_links": bool(store.get_setting(None, "hold_new_links")), + "opds_enabled": bool(store.get_setting(None, "opds_enabled")), "personal_ini": ini.read_text(encoding="utf-8") if ini.exists() else "", **{name: _download_setting(name) for name in DOWNLOAD_DEFAULTS}, **job_list_settings(), @@ -392,6 +395,7 @@ SETTINGS = ( "link_books", "hold_new_links", "pause_uploads", + "opds_enabled", "download_template", "download_date_format", *JOB_LIST_DEFAULTS, @@ -411,6 +415,9 @@ def save_settings(**changes: Any) -> dict[str, Any]: link_books, personal_ini = changes.get("link_books"), changes.get("personal_ini") if link_books is not None and not isinstance(link_books, bool): raise ValueError("Link books to CWA library must be true or false") + opds_enabled = changes.get("opds_enabled") + if opds_enabled is not None and not isinstance(opds_enabled, bool): + raise ValueError("Enable OPDS must be true or false") numbers: dict[str, int] = {} if "recent_chapters" in changes: numbers["recent_chapters"] = whole_number(changes["recent_chapters"], "The number of recent chapters") @@ -430,6 +437,8 @@ def save_settings(**changes: Any) -> dict[str, Any]: check(texts[name]) warnings = fff.check_personal_ini(personal_ini) if personal_ini is not None else [] + if opds_enabled is not None: + store.set_setting(None, "opds_enabled", opds_enabled) released = False if link_books is not None and _linking_available(): released = _link_books() != link_books @@ -664,12 +673,38 @@ def _download_setting(name: str) -> str: return store.get_setting(None, name) or DOWNLOAD_DEFAULTS[name] -def _download_name(story: dict[str, Any]) -> str: - """The file name of a downloaded book, from the download template. If the template fails or gives no name for - this book, the name is the title.""" +OPDS_PASSWORD_LENGTH = (8, 256) + + +def _opds_password_name(username: str) -> str: + return f"opds_password_hash:{username.strip().casefold()}" + + +def opds_password_hash(username: str) -> str | None: + """The hash of the OPDS password of a user in the proxy mode, or None.""" + return store.get_setting(None, _opds_password_name(username)) + + +def set_opds_password(username: str, password: Any) -> None: + """Set the OPDS password of a user. Bindery keeps only the hash.""" + least, most = OPDS_PASSWORD_LENGTH + if not isinstance(password, str) or not least <= len(password) <= most: + raise ValueError(f"The OPDS password must be text of {least} to {most} characters") + store.set_setting(None, _opds_password_name(username), generate_password_hash(password)) + + +def book_metadata(story: dict[str, Any]) -> dict[str, Any]: + """The metadata of the book of a story, with the stored title and author if the site gave none.""" meta = _book_metadata(story["site_values"], _epub_overrides(story["pinned"])) meta["title"] = meta.get("title") or story["title"] meta["authors"] = meta.get("authors") or [story["author"]] + return meta + + +def _download_name(story: dict[str, Any]) -> str: + """The file name of a downloaded book, from the download template. If the template fails or gives no name for + this book, the name is the title.""" + meta = book_metadata(story) meta.update(publisher=story["site"], timestamp=story["created_at"], last_modified=story["last_changed"]) # Without CWA, the saved CWA book IDs do not apply. book_id = story["calibre_id"] if config.CWA_ENABLED and story["calibre_id"] is not None else story["id"] diff --git a/bindery/store.py b/bindery/store.py index 9653bb9..516e2d7 100644 --- a/bindery/store.py +++ b/bindery/store.py @@ -204,6 +204,18 @@ class Store: stories = [dict(_story(r), sort_title=sort_title(r["title"] or r["url"])) for r in rows] return sorted(stories, key=lambda s: s["sort_title"]) + def catalog(self) -> list[dict[str, Any]]: + """The stories with included chapters, with the values of the OPDS catalog. A story has one cover at most, + and `cover_key` changes with it.""" + with self._connect() as conn: + rows = conn.execute(""" + SELECT s.id, s.url, s.site, s.title, s.author, s.pinned, s.site_values, s.last_changed, s.created_at, + (SELECT COUNT(*) FROM chapters c WHERE c.story_id = s.id AND c.included = 1) AS chapter_count, + i.mime AS cover_mime, i.url || ' ' || length(i.data) AS cover_key + FROM stories s LEFT JOIN images i ON i.story_id = s.id AND i.is_cover = 1 + WHERE EXISTS (SELECT 1 FROM chapters c WHERE c.story_id = s.id AND c.included = 1)""").fetchall() + return [_story(r) for r in rows] + def update_story(self, story_id: int, **fields: Any) -> None: if not fields: return diff --git a/bindery/web.py b/bindery/web.py index ab99fe2..a8d2a1d 100644 --- a/bindery/web.py +++ b/bindery/web.py @@ -2,7 +2,6 @@ import base64 import hashlib -import io import logging import os import re @@ -13,10 +12,10 @@ from datetime import timedelta from pathlib import Path from typing import Any -from flask import Flask, Response, g, jsonify, request, send_file, session +from flask import Flask, Response, g, jsonify, request, session from werkzeug.exceptions import Forbidden, HTTPException -from bindery import auth, config, orchestrator +from bindery import auth, config, opds, orchestrator from bindery.cwa import CwaError from bindery.epub import EpubError from bindery.fff import StoryError @@ -69,6 +68,7 @@ def _secret_key() -> bytes: config.prepare() app = Flask(__name__) +app.register_blueprint(opds.bp) app.config.update( SECRET_KEY=_secret_key(), SEND_FILE_MAX_AGE_DEFAULT=0, @@ -148,13 +148,27 @@ def check_host() -> Any: return jsonify({"error": f"Bindery does not answer to the host name {request.host}. Add it to ALLOWED_HOSTS."}), 400 +def _catalog_login() -> Any: + """The login of the OPDS catalog. It takes no API key. See `auth.catalog_login`.""" + if not orchestrator.store.get_setting(None, "opds_enabled"): + return jsonify({"error": "Not found"}), 404 + address = auth.client_address(request.headers, request.remote_addr) + if auth.catalog_login(request.headers, session, address, orchestrator.opds_password_hash, orchestrator.cwa): + return None + # The header makes a reader ask for the user name and password. + return jsonify({"error": "Unauthorized"}), 401, {"WWW-Authenticate": 'Basic realm="Bindery", charset="UTF-8"'} + + @app.before_request def check_login() -> Any: - if not request.path.startswith("/api/") or request.path == "/api/health": + catalog = request.path.startswith("/opds/") or request.path == "/opds" + if not (catalog or request.path.startswith("/api/")) or request.path == "/api/health": return None error = auth.config_error() if error: return jsonify({"error": f"{error}. Fix the setting and restart Bindery."}), 503 + if catalog: + return _catalog_login() identity = _identity() if ( request.method in MUTATING @@ -269,6 +283,7 @@ def app_config() -> Any: "job_list_max_hours": config.JOB_LIST_MAX_HOURS, "job_list_max_count": config.JOB_LIST_MAX_COUNT, "download_fields": orchestrator.download_fields(), + "auth_mode": config.AUTH_METHOD, **orchestrator.job_list_settings(), } @@ -278,13 +293,31 @@ def settings() -> Any: if request.method == "PUT": _admin_only() data = _json() - return orchestrator.save_settings(**{name: data.get(name) for name in orchestrator.SETTINGS}) - settings = orchestrator.get_settings() + return _with_opds_user(orchestrator.save_settings(**{name: data.get(name) for name in orchestrator.SETTINGS})) + settings = _with_opds_user(orchestrator.get_settings()) if not auth.is_admin(g.identity): del settings["personal_ini"] # it can hold passwords for story sites return settings +def _with_opds_user(settings: dict[str, Any]) -> dict[str, Any]: + """Adds the user name that the catalog expects from this user, and whether this user has an OPDS password.""" + user = g.identity.name if config.AUTH_METHOD == auth.AUTH_PROXY and g.identity.via == "proxy" else None + return {**settings, "opds_user": user, "opds_password_set": bool(user and orchestrator.opds_password_hash(user))} + + +@app.put("/api/settings/opds-password") +def opds_password() -> Any: + """Each user in the proxy mode sets their own OPDS password. The user name is the one from the proxy.""" + identity = g.identity + if config.AUTH_METHOD != auth.AUTH_PROXY or identity.via != "proxy" or not identity.name: + raise ValueError("The OPDS password is for users who log in through the proxy") + orchestrator.set_opds_password(identity.name, _json().get("password")) + # Tries before the password existed must not lock the user out of the new password. + auth.limiter.forget(identity.name) + return {"opds_user": identity.name, "opds_password_set": True} + + # Jobs @@ -399,23 +432,12 @@ def delete_chapters(story_id: int) -> Any: @app.get("/api/stories//epub") def story_epub(story_id: int) -> Any: - file, name = orchestrator.build_for_download(story_id) - response = send_file(file, download_name=name, as_attachment=True, mimetype="application/epub+zip") - response.content_length = os.fstat(file.fileno()).st_size - return response + return opds.send_epub(story_id) @app.get("/api/stories//cover") def story_cover(story_id: int) -> Any: - cover = orchestrator.story_cover(story_id) - if not cover: - raise KeyError("This story has no cover") - # The page puts the cover version in the URL, so the browser can keep the image for a year. - response = send_file(io.BytesIO(cover["data"]), mimetype=cover["mime"], max_age=365 * 24 * 60 * 60) - # A cover needs a login, so a shared cache must not keep it. - response.cache_control.public = False - response.cache_control.private = True - return response + return opds.send_cover(story_id) # Replacement rules. Story rules run after the global rules. diff --git a/tests/test_api.py b/tests/test_api.py index f0a9d1d..cfb516b 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -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, diff --git a/tests/test_opds.py b/tests/test_opds.py new file mode 100644 index 0000000..0bf7f0a --- /dev/null +++ b/tests/test_opds.py @@ -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 = "

Stored text.

" + 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 \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 " + 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": "

First part.

Second part.

", + "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"", 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}"} diff --git a/tests/test_opds_auth.py b/tests/test_opds_auth.py new file mode 100644 index 0000000..2b2ca6e --- /dev/null +++ b/tests/test_opds_auth.py @@ -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 diff --git a/tests/test_opds_settings.py b/tests/test_opds_settings.py new file mode 100644 index 0000000..5160d08 --- /dev/null +++ b/tests/test_opds_settings.py @@ -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"