13 Commits
Author SHA1 Message Date
tim d28f69d93a Show the story sort headings as chips on narrow screens 2026-10-03 11:53:47 +02:00
tim 60424e8e74 Name the publishing site in the README
"The site" could mean the publishing site of a story or the Bindery
site. Name the publishing site, its publisher and its published text.
2026-10-03 11:23:28 +02:00
tim 5ac56de09d Document the OPDS catalog
Cover the catalog, its logins in each mode, and the paths that the
proxy must let through.
2026-10-03 11:23:28 +02:00
tim 3662372c38 Add the OPDS settings to the settings page
The Book Downloads card has an "Enable OPDS" box. In the proxy mode, it
also shows the user name for the reader and a field to set the OPDS
password. The field is grey until the box is checked. Disabled text
fields now look grey.
2026-10-03 11:23:28 +02:00
tim 28cd0dcde9 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.
2026-10-03 11:23:28 +02:00
tim ece59c2558 Add {series_sort} to the download template
It gives the series with a leading article at the end, as Calibre's default does for {series}.
2026-10-03 08:09:41 +02:00
tim dcdd8e4ac7 Keep the article in {title} and {series}
The default template is now {title_sort} - {authors}. It gives the same name as Calibre's default.
2026-10-03 08:09:41 +02:00
tim 42e7287389 Fold each group of settings into an open card
The long help for the download template and the date format now folds shut.
2026-10-03 08:09:41 +02:00
tim e38c8d655a Add a date format for download names
The new setting gives the strftime format of {pubdate}, {timestamp} and {last_modified}.
The default is Calibre's "%b, %Y".
2026-10-03 08:09:41 +02:00
tim 2c8a1bb187 Use {title} - {authors} as the default download template
This is Calibre's default. An empty template now restores the default.
2026-10-03 08:09:41 +02:00
tim 323bec3c04 Name no template fields that Bindery does not have 2026-10-03 08:09:41 +02:00
tim 3e29ab6bf0 Name downloaded books with a template
The new "Book Downloads" group in Settings holds a Calibre template for the file name of a
direct download. Without a template, the name stays the title. The name is also the title if
the template fails or gives no name for a book.
2026-10-03 08:09:41 +02:00
tim 9e7a488b50 Add Calibre's file name templates
The module prepares the metadata of a book and expands a template as Calibre 9.2.1 does with
its default tweaks. It takes fields, formats, indexes, and text before and after a field. It
refuses template functions and the program modes.
2026-10-03 08:09:41 +02:00
28 changed files with 2644 additions and 90 deletions
+1
View File
@@ -28,6 +28,7 @@ The server has no reloader. If you change Python code, restart it. Flask reads `
- `bindery/fff.py` is the only module that calls FanFicFare.
- `bindery/cwa.py` talks to CWA through OPDS. Bindery never opens the CWA database.
- `bindery/diffing.py` hashes and diffs text, not HTML.
- `bindery/filename_template.py` expands Calibre's file name templates. It never runs template text as code.
- `bindery/jobs.py` keeps the job queue in SQLite, so jobs survive a restart.
- `tests/fake_cwa.py` is a fake CWA server.
+123 -19
View File
@@ -12,8 +12,8 @@ FanFicFare builds a book from the chapter list on the site, so a chapter that th
also goes from your book. **Bindery keeps every chapter that it downloads** in its own database, and
builds each ePub from that database.
- **Chapters that the site removes stay in the book.** A new chapter goes after the chapter that
comes before it on the site. For example, you have chapters 1 to 60, and the site now lists only 1
- **Chapters that the publisher removes stay in the book.** A new chapter goes after the chapter that
comes before it on the publishing site. For example, you have chapters 1 to 60, and the publishing site now lists only 1
to 4 and 61. The book then has chapters 1 to 61.
- **A changed chapter keeps its old text as a revision.** You can compare the revisions and choose
the one that goes in the book.
@@ -135,6 +135,17 @@ Each build without a saved CWA book ID then puts a plain ePub in that folder, in
or publish the port of Bindery only on the host, as `127.0.0.1:8085:8085` in the Compose file.
`TRUST_PROXY` does not change this check. It is only for the lockouts.
#### Paths without the proxy login
An OPDS reader cannot use the login page of the proxy, such as OIDC. Set the proxy to skip its login
for `/opds` and each path below it. If you set `URL_BASE`, put it before the path. As a regular
expression, this is `^/opds([/?].*)?$`. In Authelia, add an `access_control` rule with
`policy: bypass` and this expression in `resources`. In Authentik, add it to "Unauthenticated
Paths" of the proxy provider.
On these paths, Bindery checks the OPDS login itself. See "OPDS catalog". Keep the proxy login on
`/api` and on all other paths. The health check at `/api/health` needs no login.
**If the login settings are incomplete or wrong, Bindery refuses every request**, also with the API
key, and shows the reason.
@@ -234,7 +245,7 @@ In the `cwa` and `proxy` modes, `BINDERY_ADMIN_USERS` lists the users that are a
- Open the Settings page, and change the settings on it.
- Change the global replacement rules.
- Read and change `personal.ini`, which can hold passwords for story sites.
- Read and change `personal.ini`, which can hold passwords for publishing sites.
Each other user can still add, update and delete stories, and change the settings and the rules of a
story. Bindery ignores the case of the names.
@@ -298,6 +309,39 @@ The answer lists the queued jobs and the unsupported URLs: `{"jobs": [...], "err
If Bindery runs under `URL_BASE`, put the prefix before `/api`.
### OPDS catalog
OPDS lets a reader app browse and download books from Bindery. An admin turns it on with "Enable OPDS"
under Settings > Book Downloads. The catalog URL is `http://localhost:8085/opds`, with your server
address in place of `localhost`. If Bindery runs under `URL_BASE`, put it before `/opds`.
The catalog has three lists, with 50 books per page:
- By Title, in title order without a leading "The", "A" or "An".
- By Author, in surname order. Each author leads to their books. A book shows under each of its authors.
- By Last Changed, with the newest change first.
A book appears when it has a stored, included chapter. It shows its authors, series, description, tags,
language and cover, with your kept values in place of the published values. A reader can search by title,
author, series and tag. A download uses the stored chapters, the current ePub rules and the download
template for the file name.
`AUTH_METHOD` sets how a reader logs in:
- `none`: The reader needs no login.
- `local`: The reader sends your Bindery user name and password with HTTP Basic authentication.
A browser with a Bindery session works too.
- `cwa`: The reader sends your CWA user name and password with HTTP Basic authentication.
A browser with a Bindery session works too.
- `proxy`: Each user sets an OPDS password under Settings > Book Downloads. Settings shows the user
name for the reader, which is the name in `PROXY_AUTH_USER_HEADER`. The reader sends that name and
password with HTTP Basic authentication. The password has 8 to 256 characters, and Bindery keeps a
hash of it. Set the proxy to skip its login for the catalog. See "Paths without the proxy login".
Bindery keeps a good login for five minutes. Wrong passwords count toward the lockouts in
"Lockouts and sessions". With `TRUST_PROXY=true`, each reader has its own lockout.
Use HTTPS outside a trusted local network, because HTTP Basic sends the password with each request.
### FanFicFare settings
On first start, Bindery copies `data/personal.ini` to `/config/personal.ini`. It uses the format of
@@ -308,13 +352,13 @@ explains the format, and
setting. Bindery reads the file for each job, so a change applies to the next job.
You can edit the file on the Settings page. Bindery refuses a file that FanFicFare cannot read, and
shows FanFicFare's warnings for other problems. **The file can hold passwords for story sites, and
shows FanFicFare's warnings for other problems. **The file can hold passwords for publishing sites, and
each admin can read them.** If `BINDERY_ADMIN_USERS` is not set, each user is an admin. See "Admins".
Bindery always sets `always_reload_first_chapter`, `mark_new_chapters`, `continue_on_chapter_error`,
`make_directories`, `zip_output` and `calibre_series_meta`, because it needs fixed values for them.
**Each book gets the cover from the site.** FanFicFare downloads the cover only if `include_images`
**Each book gets the cover from the publisher.** FanFicFare downloads the cover only if `include_images`
is on. If it is off, Bindery sets it to `coveronly`, which keeps the cover but leaves out the images
in the story text.
@@ -417,7 +461,7 @@ Make a CWA user for Bindery with these settings:
- The download permission. Bindery reads story URLs and chapters from the ePubs.
- "Show Publisher Section", in the sidebar settings of the user. CWA shows its list of publishers in
OPDS only to users with this setting. "Import from the library" uses the list to find the books of
story sites, because FanFicFare sets the publisher of a book to the site name.
publishing sites, because FanFicFare sets the publisher of a book to the site name.
- Access to all books. Bindery finds only the books that this account can see.
Bindery sends the password of this account with each OPDS request. **Use a password that no other
@@ -428,7 +472,7 @@ unset, and give the guest user of CWA the settings above.
## Import an ePub file
If the site removed chapters before Bindery tracked the story, import an ePub that you made earlier
If the publishing site removed chapters before Bindery tracked the story, import an ePub that you made earlier
with FanFicFare. Bindery stores its chapters as if it downloaded them. Chapters that the site
publishes later go after them.
@@ -441,18 +485,78 @@ publishes later go after them.
that FanFicFare writes into each ePub. A chapter that Bindery stored already keeps its stored text.
The other chapters go in the order of the ePub.
After the import, Bindery checks the site for new chapters and sends the book to CWA. If you clear
After the import, Bindery checks the publishing site for new chapters and sends the book to CWA. If you clear
"Check the site for new chapters after import", Bindery only rebuilds the book of a story that it
downloaded before.
## File names of downloaded books
"Download ePub" on a story page and each download from the OPDS catalog send the book with a file
name. A template gives the name. The default template is `{title_sort} - {authors}`. For "The Book"
by Ann Writer, it gives `Book, The - Ann Writer.epub`. That is the name that Calibre's default template gives.
An admin can change the template under "Book Downloads" in Settings. To go back to the default,
clear the field and save. The template follows
the rules of [Calibre's templates](https://manual.calibre-ebook.com/template_lang.html) for file
names. Bindery adds `.epub`. For example, `{series:|| - }{series_index:0>2s|| - }{title} - {authors}`
gives `The Saga - 02 - The Book - Ann Writer.epub`.
In Calibre, `{title}` and `{series}` give the title sort in a file name, for example `Book, The`.
In Bindery, they keep the article. This is the same as Calibre with the tweak
`save_template_title_series_sorting` set to `strictly_alphabetic`. Use `{title_sort}` and
`{series_sort}` for the title and the series with the article at the end. `{series_sort}` is a
Bindery field. Calibre does not have it, and gives empty text for it.
| Field | Value |
| --- | --- |
| `{title}` | The title: `The Book`. |
| `{title_sort}` | The title with a leading "The", "A" or "An" at the end: `Book, The`. |
| `{authors}`, `{author}` | The authors, with ` & ` between them. |
| `{author_sort}` | The authors with the last name first: `Writer, Ann`. |
| `{series}` | The series. |
| `{series_sort}` | The series with a leading article at the end: `Saga, The`. |
| `{series_index}` | The series number: `2`, or `2.5`. |
| `{tags}` | The tags in sorted order, with `, ` between them. |
| `{languages}` | The language, as Bindery stores it. |
| `{publisher}` | The publishing site. |
| `{pubdate}` | The published date. |
| `{timestamp}` | The date when Bindery added the story. |
| `{last_modified}` | The date when the chapters last changed. |
| `{id}` | The CWA book ID. For a story with no library book, and without CWA, the Bindery story ID. |
- A field with no value becomes empty text.
- `{field:format|text before|text after}` adds the two texts only when the field has a value. Each
of the three parts can be empty. `{series:|| - }` adds ` - ` after the series.
- The format is a Python format. `{series_index:0>3s}` gives `002`, and `{title:.20s}` gives the
first 20 characters. `{author_sort[0]}` gives the first character.
- `{{` and `}}` give braces. Bindery makes each run of spaces one space.
- A download has one file name. If the template has folders, Bindery uses the part after the last
`/`. That is the name that Calibre gives the file.
- Bindery does not support template functions, such as `{title:uppercase()}`, or the program modes
of Calibre. It refuses such a template when you save it.
- If the template fails for a book, or gives an empty name, Bindery names the file after the title.
The log has a warning.
- Bindery replaces each character that a file name cannot have with `_`. A name has 100 characters
and 200 bytes at most.
"Date format" sets the form of `{pubdate}`, `{timestamp}` and `{last_modified}`. It takes the
[codes of Python's strftime](https://docs.python.org/3/library/datetime.html#format-codes). `%d`
gives the day, `%b` the month name, `%m` the month number and `%Y` the year. The default is
`%b, %Y`, as in Calibre. It gives `May, 2020`. To go back to the default, clear the field and save.
A `/` in a date becomes `_`, as in Calibre.
Bindery follows Calibre 9.2.1 for English, with the tweak above. The order of tags with letters
outside ASCII can differ from Calibre. The template applies only to downloads in the browser. It
does not change the names of the files in `INGEST_DIR`. The browser can ignore the name.
## Download modes
| Mode | What it does |
| --- | --- |
| Download new chapters | Downloads the chapters that are not stored yet. The default, and what scheduled updates use. |
| Recheck selected chapters | Downloads the selected chapters again. Changed text becomes a new revision. |
| Recheck all chapters | Downloads every chapter on the site again, and new chapters too. |
| Download and replace entire book | Downloads every chapter again. Chapters that the site no longer lists leave the book. They stay stored, and you can put them back. |
| Recheck all chapters | Downloads every chapter on the publishing site again, and new chapters too. |
| Download and replace entire book | Downloads every chapter again. Chapters that the publishing site no longer lists leave the book. They stay stored, and you can put them back. |
| Update metadata from site | Updates only the metadata in the ePub. |
| Rebuild book | Builds the book again from the stored chapters, and sends it. |
@@ -472,26 +576,26 @@ recheck, or use one of these settings:
story. This recheck takes the place of the next scheduled update. The story then ignores the "last
N chapters" setting.
Each rechecked chapter is one request to the site, so keep the numbers small for long stories.
Each rechecked chapter is one request to the publishing site, so keep the numbers small for long stories.
If the site changes the title of a chapter, "Download new chapters" and "Recheck selected chapters"
If the publishing site changes the title of a chapter, "Download new chapters" and "Recheck selected chapters"
download that chapter again. The chapter gets the new title only if the download works. On sites
that number chapter URLs, such as FanFiction.net, this also finds chapters that moved when the
author inserted a chapter.
### Revisions
A recheck compares the text of a chapter with its newest revision, which is the text that the site
sent last. Bindery stores the HTML as the site sends it, before the rules, so a rule never makes a
A recheck compares the text of a chapter with its newest revision, which is the text that the publishing site
sent last. Bindery stores the HTML as the publishing site sends it, before the rules, so a rule never makes a
chapter look changed.
- If the text changed, the new revision goes in the book. If you chose an older revision, it
stays in the book until the site changes the chapter again.
stays in the book until the publishing site changes the chapter again.
- On the page of a chapter, you can compare revisions, choose the one that goes in the book, and
delete the others. The revision in the book and the newest revision cannot be deleted.
- A chapter imported from an ePub holds its text after the FanFicFare settings and rules of that
build. If the current rules turn the site text into the imported text, Bindery counts the chapter
as unchanged, and stores the site text in its place. Other differences, for example from settings
build. If the current rules turn the published text into the imported text, Bindery counts the chapter
as unchanged, and stores the published text in its place. Other differences, for example from settings
in `personal.ini`, show as a change.
### Restarts
@@ -507,7 +611,7 @@ Bindery to start the thread again.
### Backups
Once a day, Bindery copies its database and `personal.ini` to `/config/backups`, and keeps the
copies of the last 7 days. **The database is the only copy of the chapters that sites removed.** The
copies of the last 7 days. **The database is the only copy of the chapters that publishing sites removed.** The
backups are on the same disk, so also copy `/config` to another disk or machine.
To restore a backup, do these steps:
@@ -524,7 +628,7 @@ Bindery keeps the metadata of each story in its database, and writes it into eac
are title, authors, description, tags, series, published date and language. Each field is locked or
unlocked for each story:
- An unlocked field takes the site value on every update.
- An unlocked field takes the published value on every update.
- A locked field keeps its stored value. The first download sets that value. For a story that you
import from the library or from an ePub file, the value comes from that ePub.
- To update a locked field once, select it under "Also take these locked or edited fields from the
+122 -12
View File
@@ -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))
+321
View File
@@ -0,0 +1,321 @@
"""Calibre's templates for file names, for the name of a downloaded book.
Bindery follows Calibre 9.2.1 with its default tweaks and the English articles. prepare() takes the tweak for
the order of titles and series as an argument. Dates take a strftime format, with Calibre's "%b, %Y" as the
default. Bindery takes {field}, {field[0]}, {field:format|prefix|suffix} and {{ }}. It refuses template
functions and the program modes. It never runs template text as code.
{series_sort} is a Bindery field. It is to {series} what {title_sort} is to {title}. Calibre has no such field.
The tags sort close to Calibre's ICU rules, but their order can differ for characters outside ASCII.
"""
import re
import unicodedata
from datetime import datetime
from string import Formatter
from typing import Any
from bindery.ingest import sanitize_filename
# The name that Calibre's default template, {title} - {authors}, gives with its default tweaks.
DEFAULT_TEMPLATE = "{title_sort} - {authors}"
DEFAULT_DATE_FORMAT = "%b, %Y"
MAX_TEMPLATE_LENGTH = 1024
MAX_DATE_FORMAT_LENGTH = 100
# The most characters that a format can ask for, as a width or a precision.
MAX_WIDTH = 255
MAX_NAME_BYTES = 200
# The fields that Bindery has, with the help text for each. Any other field is empty, as it is in Calibre for
# a book without that value.
FIELDS = {
"title": "The title",
"title_sort": 'The title with a leading "The", "A" or "An" at the end, such as "Book, The"',
"authors": "The authors",
"author": "The authors",
"author_sort": "The author sort string. To use only the first letter of the name, use {author_sort[0]}",
"series": "The series",
"series_sort": 'The series with a leading "The", "A" or "An" at the end, such as "Saga, The". '
"Calibre does not have this field",
"series_index": "The series number. To get leading zeros, use {series_index:0>3s}",
"tags": "The tags",
"languages": "The language",
"publisher": "The story site",
"pubdate": "The published date",
"timestamp": "The date when Bindery added the story",
"last_modified": "The date when the chapters last changed",
"id": "The Bindery story ID",
}
class TemplateError(ValueError):
"""A template that Bindery cannot use, or that fails for a book."""
# Metadata, as Calibre prepares it for a file name
_ARTICLE = re.compile(r"^(A\s+|The\s+|An\s+)", re.IGNORECASE)
# Calibre's quote_pairs: each opening quote, with the quotes that close it.
_QUOTE_PAIRS = {
'"': '"',
"'": "'",
"\u201c": "\u201d\u201c",
"\u201d": "\u201d",
"\u201e": "\u201d\u201c",
"\u201a": "\u2019\u2018",
"\u2019": "\u2019\u2018",
"\u2018": "\u2019\u2018",
"\u2039": "\u203a",
"\u203a": "\u2039",
"\u300a": "\u300b",
"\u3008": "\u3009",
"\u00bb": "\u00ab\u00bb",
"\u00ab": "\u00ab\u00bb",
"\u300c": "\u300d",
"\u300e": "\u300f",
}
# The ASCII punctuation in the order of Calibre's sort. Digits come next, then letters.
_PUNCTUATION = " _-,;:!?.'\"()[]{}@*/\\&#%`^+<=>|~$"
# Calibre's default tweaks for author names.
_NAME_PREFIXES = {"mr", "mrs", "ms", "dr", "prof"}
_NAME_SUFFIXES = {"jr", "sr", "inc", "ph.d", "phd", "md", "m.d", "i", "ii", "iii", "iv", "junior", "senior"}
_COPY_WORDS = {
"agency",
"corporation",
"company",
"co.",
"council",
"committee",
"inc.",
"institute",
"national",
"society",
"club",
"team",
"software",
"games",
"entertainment",
"media",
"studios",
}
_BRACKETS = {"(": ")", "[": "]", "{": "}"}
def _unquote(title: str) -> str:
closers = _QUOTE_PAIRS.get(title[:1])
if closers:
title = title[1:]
if title and title[-1] in closers:
title = title[:-1]
return title
def title_sort(title: str) -> str:
"""Calibre's title_sort(): "The Book" gives "Book, The"."""
title = _unquote(title.strip())
match = _ARTICLE.search(title)
if match:
title = _unquote(f"{title[match.end() :]}, {match.group(1)}")
return title.strip()
def _without_brackets(text: str) -> str:
depth = dict.fromkeys(_BRACKETS, 0)
closers = {close: opener for opener, close in _BRACKETS.items()}
kept = []
for char in text:
if char in depth:
depth[char] += 1
elif char in closers:
depth[closers[char]] = max(0, depth[closers[char]] - 1)
elif not any(depth.values()):
kept.append(char)
return "".join(kept)
def author_sort(author: str) -> str:
"""Calibre's author_to_author_sort() with its default tweaks: "Ann Writer" gives "Writer, Ann"."""
plain = _without_brackets(author).strip()
tokens = plain.split()
lower = [t.lower().removesuffix(".") for t in tokens]
if "," in plain or len(tokens) < 2 or _COPY_WORDS.intersection(t.lower() for t in tokens):
return author
first = next((n for n, t in enumerate(lower) if t not in _NAME_PREFIXES), None)
if first is None:
return author
last = next((n for n in range(len(tokens) - 1, first - 1, -1) if lower[n] not in _NAME_SUFFIXES), None)
if last is None:
return author
names = [tokens[last], *tokens[first:last]]
if len(names) > 1:
names[0] += ","
return " ".join(names + tokens[last + 1 :])
def series_index(index: Any) -> str:
"""Calibre's fmt_sidx(): 2.0 gives "2", and 2.345 gives "2.35"."""
try:
index = float(index)
except (TypeError, ValueError):
index = 1.0
if index.is_integer():
return str(int(index))
return f"{index:.2f}".rstrip("0")
def check_date_format(date_format: str) -> None:
"""Raise TemplateError if Bindery cannot use the date format."""
if len(date_format) > MAX_DATE_FORMAT_LENGTH:
raise TemplateError(f"A date format can have {MAX_DATE_FORMAT_LENGTH} characters at most")
try:
datetime(2020, 5, 6).strftime(date_format)
except ValueError as e:
raise TemplateError(f"The date format is not valid: {e}") from e
def _date(value: Any, date_format: str) -> str:
try:
return datetime.fromisoformat(value).strftime(date_format)
except (TypeError, ValueError):
return ""
def _tag_key(tag: str) -> tuple[list[tuple[int, str]], str]:
"""Close to the key of Calibre's sort: no case, punctuation before digits before letters, accents last."""
text = unicodedata.normalize("NFKD", tag.casefold())
ranks = len(_PUNCTUATION)
letters = [
(_PUNCTUATION.index(c), "") if c in _PUNCTUATION else (ranks + c.isalpha(), c)
for c in text
if not unicodedata.combining(c)
]
return letters, text
def prepare(
meta: dict[str, Any], book_id: int, order: str = "library_order", date_format: str = DEFAULT_DATE_FORMAT
) -> dict[str, Any]:
"""Calibre's get_component_metadata(): the values of a book in the form that a file name takes.
`order` is Calibre's tweak save_template_title_series_sorting. With "strictly_alphabetic", the title and the
series keep a leading article. `date_format` is Calibre's timefmt.
"""
library = order == "library_order"
values: dict[str, Any] = {"id": str(book_id), "publisher": meta.get("publisher") or ""}
title, authors, series = meta.get("title"), [a for a in meta.get("authors") or [] if a], meta.get("series")
if title:
values["title_sort"] = title_sort(title)
values["title"] = values["title_sort"] if library else title
if authors:
values["authors"] = values["author"] = " & ".join(a.replace("&", "&&") for a in authors)
values["author_sort"] = " & ".join(author_sort(a) for a in authors)
if series:
values["series_sort"] = title_sort(series)
values["series"] = values["series_sort"] if library else series.strip()
if meta.get("series_index") is not None:
values["series_index"] = series_index(meta["series_index"])
values["tags"] = ", ".join(sorted(meta.get("tags") or [], key=_tag_key)).removeprefix("/")
values["languages"] = meta.get("languages") or ""
for name in ("pubdate", "timestamp", "last_modified"):
values[name] = _date(meta.get(name), date_format)
return values
# Templates
_FIELD = re.compile(r"([^.\[\]]*)((?:\[\d+\])*)")
_AFFIXES = re.compile(r"^(.*)\|([^|]*)\|(.*)$", re.DOTALL)
_INT_CODES, _FLOAT_CODES = "bcdoxXn", "eEfFgG%"
def _format(value: str, fmt: str) -> str:
"""Calibre's _do_format(): a number code takes the text as a number."""
if not fmt or not value:
return value
code = fmt[-1]
try:
number = int(value) if code in _INT_CODES else float(value) if code in _FLOAT_CODES else value
return format(number, fmt)
except (ValueError, OverflowError) as e:
raise TemplateError(f"The format {fmt!r} cannot take the value {value!r}: {e}") from e
def _parse(template: str) -> list[tuple[str, str | None, list[int], str, str, str]]:
"""The parts of a template. Each part is the text before a field, then the name of the field, its indexes,
its format, its prefix and its suffix. The name is None for text with no field after it."""
if len(template) > MAX_TEMPLATE_LENGTH:
raise TemplateError(f"A template can have {MAX_TEMPLATE_LENGTH} characters at most")
if template.startswith(("program:", "python:")):
raise TemplateError("Bindery does not support the program modes of Calibre")
try:
fields = list(Formatter().parse(template))
except ValueError as e:
raise TemplateError(f"The braces of the template do not match: {e}") from e
parts = []
for text, name, spec, conversion in fields:
if name is None:
parts.append((text, None, [], "", "", ""))
continue
field = _FIELD.fullmatch(name)
if not field or conversion or "{" in spec:
raise TemplateError(f"Bindery does not support the field {{{name}}} in this form")
if not field.group(1) or field.group(1).isdigit():
raise TemplateError("A field needs a name, such as {title}")
affixes = _AFFIXES.match(spec)
fmt, prefix, suffix = affixes.groups() if affixes else (spec, "", "")
if fmt.startswith("'") or (":'" in fmt and fmt.endswith("'")) or ("(" in fmt and fmt.endswith(")")):
raise TemplateError("Bindery does not support template functions, such as {title:uppercase()}")
if any(int(n) > MAX_WIDTH for n in re.findall(r"\d+", fmt)):
raise TemplateError(f"A format can ask for {MAX_WIDTH} characters at most")
_format("1", fmt)
indexes = [int(n) for n in re.findall(r"\d+", field.group(2))]
parts.append((text, field.group(1).lower(), indexes, fmt, prefix, suffix))
return parts
def check(template: str) -> None:
"""Raise TemplateError if Bindery cannot use the template."""
_parse(template)
def expand(template: str, values: dict[str, Any]) -> str:
"""The text of a template for the prepared values of a book, as Calibre gives it before it makes a path."""
out = []
for text, name, indexes, fmt, prefix, suffix in _parse(template):
out.append(text)
if name is None:
continue
value = values.get(name, "")
if isinstance(value, list):
value = ",".join(value)
value = value.replace("/", "_").replace("\\", "_")
try:
for index in indexes:
value = value[index]
except IndexError as e:
raise TemplateError(f"The value of {{{name}}} is too short for [{index}]") from e
if value:
value = _format(value, fmt)
if value:
out += prefix, value, suffix
return re.sub(r"\s+", " ", "".join(out)).strip(" ")
_RESERVED = re.compile(r"(?:CON|PRN|AUX|NUL|COM\d|LPT\d)(?:\.|$)", re.IGNORECASE)
def file_name(template: str, values: dict[str, Any]) -> str:
"""The name of one file from a template, without the extension. Empty if the template gives no name.
A download has one name, so Bindery uses the last part of a template with folders. That is the name that
Calibre gives the file. Windows cannot use some names, such as CON. Those get a leading underscore.
"""
parts = [part.strip() for part in expand(template, values).split("/")]
name = next((part for part in reversed(parts) if part), "")
if not name:
return ""
name = sanitize_filename(name)
if _RESERVED.match(name):
name = f"_{name}"
return name.encode()[:MAX_NAME_BYTES].decode(errors="ignore").rstrip(" .")
+297
View File
@@ -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/<order>")
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/<int:story_id>.epub")
def download(story_id: int) -> Response:
return send_epub(story_id)
@bp.get("/opds/books/<int:story_id>/cover")
def cover(story_id: int) -> Response:
return send_cover(story_id)
+104 -6
View File
@@ -11,13 +11,15 @@ from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, BinaryIO
from bindery import config, fff
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
from bindery.diffing import diff_html, text_hash
from bindery.epub import EpubError, epub_metadata, epub_urls
from bindery.fff import StoryError
from bindery.ingest import IngestError, IngestFolder
from bindery.ingest import IngestError, IngestFolder, sanitize_filename
from bindery.jobs import METADATA_FIELDS, DownloadMode, Job, QueueStatus, job_queue
from bindery.merge import merge_chapter_lists
from bindery.rules import RuleError, Step, apply_rules, compile_rules, normalize_rules
@@ -379,13 +381,25 @@ 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(),
}
# The settings that save_settings takes.
SETTINGS = ("recent_chapters", "personal_ini", "link_books", "hold_new_links", "pause_uploads", *JOB_LIST_DEFAULTS)
SETTINGS = (
"recent_chapters",
"personal_ini",
"link_books",
"hold_new_links",
"pause_uploads",
"opds_enabled",
"download_template",
"download_date_format",
*JOB_LIST_DEFAULTS,
)
def save_settings(**changes: Any) -> dict[str, Any]:
@@ -401,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")
@@ -408,8 +425,20 @@ def save_settings(**changes: Any) -> dict[str, Any]:
if name in changes:
most = config.JOB_LIST_MAX_HOURS if name.endswith("hours") else config.JOB_LIST_MAX_COUNT
numbers[name] = whole_number(changes[name], name, most)
texts: dict[str, str] = {}
for name, label, check in (
("download_template", "The download template", filename_template.check),
("download_date_format", "The date format", filename_template.check_date_format),
):
if name in changes:
if not isinstance(changes[name], str):
raise ValueError(f"{label} must be text")
texts[name] = changes[name].strip()
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
@@ -422,6 +451,9 @@ def save_settings(**changes: Any) -> dict[str, Any]:
store.set_setting(None, name, True if value else None)
for name, number in numbers.items():
store.set_setting(None, name, number)
# An empty value restores the default.
for name, text in texts.items():
store.set_setting(None, name, text or None)
if personal_ini is not None:
# Write a new file, then replace the old one, so a job never reads half a file.
partial = config.PERSONAL_INI.with_name("personal.ini.part")
@@ -623,15 +655,81 @@ def library_candidates() -> list[dict[str, Any]]:
return [b.to_dict() for b in cwa().site_books(fff.is_supported_site) if b.id not in tracked_ids]
def download_fields() -> dict[str, str]:
"""The fields of the download template, with the help text for each."""
fields = dict(filename_template.FIELDS)
if config.CWA_ENABLED:
fields["id"] = "The CWA book ID. For a story with no library book, the Bindery story ID"
return fields
DOWNLOAD_DEFAULTS = {
"download_template": filename_template.DEFAULT_TEMPLATE,
"download_date_format": filename_template.DEFAULT_DATE_FORMAT,
}
def _download_setting(name: str) -> str:
return store.get_setting(None, name) or DOWNLOAD_DEFAULTS[name]
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"]
name = ""
try:
# {title} and {series} keep a leading article, as in Calibre with this value of its tweak
# save_template_title_series_sorting. {title_sort} moves it to the end.
date_format = _download_setting("download_date_format")
values = filename_template.prepare(meta, book_id, order="strictly_alphabetic", date_format=date_format)
name = filename_template.file_name(_download_setting("download_template"), values)
except filename_template.TemplateError as e:
logger.warning(f"The download template fails for {story['url']}: {e}")
return f"{name or sanitize_filename(story['title'])}.epub"
def build_for_download(story_id: int) -> tuple[BinaryIO, str]:
"""The epub of a story for a browser download, as an open file. Bindery builds it only if the book changed
after the last build."""
"""The epub of a story for a browser download, as an open file with its file name. Bindery builds it only if
the book changed after the last build."""
story = _require_story(story_id)
overrides = _epub_overrides(story["pinned"])
key = _build_key(story, overrides, story["site_values"])
name = _download_name(story)
# The file is opened under the lock, so a new build cannot delete it first.
with _build_lock:
return open(_built_epub(story, overrides, key, record=False), "rb"), story["title"]
return open(_built_epub(story, overrides, key, record=False), "rb"), name
# Metadata locks
+3 -3
View File
@@ -227,9 +227,9 @@ export function globalRulesCard() {
.catch(showError);
return h(
'section',
{ class: 'card' },
h('h2', {}, 'Global replacement rules'),
'details',
{ class: 'card', open: true },
h('summary', {}, h('h2', {}, 'Global replacement rules')),
h(
'p',
{ class: 'hint' },
+219 -16
View File
@@ -1,10 +1,26 @@
// Settings: the recheck of recent chapters, the uploads to CWA, the global rules, and FanFicFare's personal.ini.
// Settings: the recheck of recent chapters, the uploads to CWA, the names of downloaded books, the global rules,
// and FanFicFare's personal.ini.
import { api, button, checkbox, externalLink, fill, h, showError, toast } from './lib.js';
import { globalRulesCard } from './rules.js';
const INI_DOCS = 'https://github.com/JimmXinu/FanFicFare/wiki/INI-File';
const DEFAULTS_INI = 'https://github.com/JimmXinu/FanFicFare/blob/main/fanficfare/defaults.ini';
const TEMPLATE_DOCS = 'https://manual.calibre-ebook.com/template_lang.html';
const DATE_CODES = 'https://docs.python.org/3/library/datetime.html#format-codes';
const code = (text) => h('code', {}, text);
// How a reader logs in to the catalog, for each login mode.
const OPDS_LOGIN = {
none: 'Bindery asks for no login, so the catalog needs none.',
local: 'The reader sends your Bindery user name and password with HTTP Basic authentication.',
cwa: 'The reader sends your CWA user name and password with HTTP Basic authentication.',
proxy:
'The proxy skips its login for the catalog. Each user sets an OPDS password below. The reader sends that ' +
'password and the user name from the proxy with HTTP Basic authentication.',
};
// A card of settings that folds shut. It starts open.
const card = (title, ...children) =>
h('details', { class: 'card', open: true }, h('summary', {}, h('h2', {}, title)), ...children);
export function settingsPage(main, _params, config) {
document.title = 'Settings · Bindery';
@@ -35,6 +51,63 @@ export function settingsPage(main, _params, config) {
pauseUploads.setAttribute('aria-describedby', pauseUploads.disabled ? 'ingest-note' : '');
ingestNote.hidden = !pauseUploads.disabled;
};
const opdsEnabled = checkbox(false, (checked) => showPassword(passwordSet, checked), {
id: 'opds-enabled',
'aria-describedby': 'opds-note',
});
let passwordSet = false;
const opdsUser = h('input', {
id: 'opds-user',
type: 'text',
readonly: true,
spellcheck: 'false',
autocomplete: 'off',
'aria-describedby': 'opds-password-status',
});
const opdsPassword = h('input', {
id: 'opds-password',
type: 'password',
minlength: 8,
maxlength: 256,
autocomplete: 'new-password',
});
const opdsPasswordStatus = h('p', { class: 'hint', id: 'opds-password-status' });
const setPassword = button('Set password', async () => {
await api('settings/opds-password', 'PUT', { password: opdsPassword.value });
opdsPassword.value = '';
showPassword(true, opdsEnabled.checked);
toast('Saved');
});
// The password applies only when the catalog is on, so the field is grey while the box is not checked.
const showPassword = (isSet, enabled) => {
passwordSet = isSet;
opdsPassword.disabled = !enabled;
setPassword.disabled = !enabled;
setPassword.textContent = isSet ? 'Reset password' : 'Set password';
fill(
opdsPasswordStatus,
isSet
? 'You have an OPDS password. Bindery keeps only a hash of it. ' +
'Enter the user name and the password in your reader.'
: 'You have no OPDS password yet. Set one, then enter the user name and the password in your reader.',
);
};
const template = h('input', {
id: 'download-template',
class: 'template',
type: 'text',
maxlength: 1024,
spellcheck: 'false',
autocomplete: 'off',
});
const dateFormat = h('input', {
id: 'download-date-format',
class: 'template',
type: 'text',
maxlength: 100,
spellcheck: 'false',
autocomplete: 'off',
});
const hoursField = (id) => h('input', { id, type: 'number', min: 0, max: config.job_list_max_hours, step: 1 });
const countField = (id) => h('input', { id, type: 'number', min: 0, max: config.job_list_max_count, step: 1 });
const storyJobsHours = hoursField('story-jobs-hours');
@@ -73,6 +146,11 @@ export function settingsPage(main, _params, config) {
recent.value = settings.recent_chapters;
showUploadSettings(settings);
ini.value = settings.personal_ini;
opdsEnabled.checked = settings.opds_enabled ?? false;
opdsUser.value = settings.opds_user ?? '';
showPassword(settings.opds_password_set ?? false, opdsEnabled.checked);
template.value = settings.download_template;
dateFormat.value = settings.download_date_format;
storyJobsHours.value = settings.story_jobs_hours;
storyJobsCount.value = settings.story_jobs_count;
mainJobsHours.value = settings.main_jobs_hours;
@@ -82,11 +160,9 @@ export function settingsPage(main, _params, config) {
fill(
main,
h(
'section',
{ class: 'card' },
h('h1', {}, 'Settings'),
h('h2', {}, 'Recheck recent chapters'),
card(
'Recheck recent chapters',
h(
'div',
{ class: 'controls' },
@@ -102,10 +178,8 @@ export function settingsPage(main, _params, config) {
),
saveButton(() => ({ recent_chapters: recent.value === '' ? 0 : Number(recent.value) })),
),
config.cwa_enabled && h(
'section',
{ class: 'card' },
h('h2', {}, 'Uploads to CWA'),
config.cwa_enabled && card(
'Uploads to CWA',
h('label', {}, linkBooks, ' Link books to CWA library'),
h(
'p',
@@ -139,10 +213,141 @@ export function settingsPage(main, _params, config) {
},
),
),
card(
'Book Downloads',
h('label', {}, opdsEnabled, ' Enable OPDS'),
h(
'section',
{ class: 'card' },
h('h2', {}, 'Jobs list'),
'p',
{ class: 'hint', id: 'opds-note' },
'OPDS lets a reader browse and download books. It is off by default. After you enable it and save, use ',
h('a', { href: 'opds' }, 'the OPDS catalog'),
'. ',
OPDS_LOGIN[config.auth_mode] ?? OPDS_LOGIN.local,
' Use HTTPS outside a trusted local network.',
),
config.auth_mode === 'proxy' && [
h(
'div',
{ class: 'credentials' },
h('label', { for: 'opds-user' }, 'OPDS user name'),
opdsUser,
h('label', { for: 'opds-password' }, 'OPDS password'),
h('div', { class: 'controls' }, opdsPassword, setPassword),
),
opdsPasswordStatus,
],
h('div', { class: 'controls' }, h('label', { for: 'download-template' }, 'File name template'), template),
h(
'p',
{ class: 'hint' },
'This template gives the file name of a book that you download from a story page or OPDS. ' +
'To go back to the default, ',
code('{title_sort} - {authors}'),
', clear the field and save.',
),
h(
'details',
{ class: 'help' },
h('summary', {}, 'File name template help'),
h(
'p',
{ class: 'hint' },
'The template follows the ',
externalLink(TEMPLATE_DOCS, 'template rules of Calibre'),
' for file names. Bindery replaces each field in braces with the value of the book, and adds .epub. ' +
'A field with no value becomes empty text. For example, ',
code('{series:|| - }{series_index:0>2s|| - }{title} - {authors}'),
' gives "The Saga - 02 - The Book - Ann Writer.epub".',
),
h('p', { class: 'hint' }, 'You can use these fields:'),
h(
'dl',
{ class: 'fields' },
Object.entries(config.download_fields).map(([name, text]) => [h('dt', {}, code(`{${name}}`)), h('dd', {}, text)]),
),
h(
'ul',
{ class: 'hint' },
h('li', {}, code('{title:.20s}'), ' gives the first 20 characters of the title.'),
h('li', {}, code('{author_sort[0]}'), ' gives the first character of the author sort string.'),
h(
'li',
{},
code('{series:|| - }'),
' adds " - " after the series, but only for a book with a series. The form is ',
code('{field:format|text before|text after}'),
'.',
),
h('li', {}, code('{{'), ' and ', code('}}'), ' give braces.'),
h(
'li',
{},
'{title} and {series} keep a leading "The", "A" or "An". {title_sort} and {series_sort} move it to the ' +
'end, so ' +
'"The Book" becomes "Book, The". Calibre puts the title sort in {title} as well, unless you set its ' +
'tweak save_template_title_series_sorting to strictly_alphabetic.',
),
h(
'li',
{},
'A download has one file name. If the template has folders, Bindery uses the part after the last ',
code('/'),
'.',
),
h(
'li',
{},
'Bindery does not support template functions, such as ',
code('{title:uppercase()}'),
', or the program modes of Calibre.',
),
h('li', {}, 'If the template fails or gives no name for a book, Bindery names the file after the title.'),
),
),
h('div', { class: 'controls' }, h('label', { for: 'download-date-format' }, 'Date format'), dateFormat),
h(
'p',
{ class: 'hint' },
'The form of template dates in {pubdate}, {timestamp} and {last_modified}. To go back to the default, ',
code('%b, %Y'),
', clear the field and save.',
),
h(
'details',
{ class: 'help' },
h('summary', {}, 'Date format help'),
h(
'p',
{ class: 'hint' },
code('%d'),
' gives the day, ',
code('%b'),
' the month name, ',
code('%m'),
' the month number and ',
code('%Y'),
' the year. See the ',
externalLink(DATE_CODES, 'list of all codes'),
'. The default gives "May, 2020". A / in a date becomes _, as in Calibre.',
),
),
saveButton(
() => ({
opds_enabled: opdsEnabled.checked,
download_template: template.value,
download_date_format: dateFormat.value,
}),
(settings) => {
opdsEnabled.checked = settings.opds_enabled ?? false;
showPassword(passwordSet, opdsEnabled.checked);
template.value = settings.download_template;
dateFormat.value = settings.download_date_format;
toast('Saved');
},
),
),
card(
'Jobs list',
h(
'div',
{ class: 'controls' },
@@ -175,10 +380,8 @@ export function settingsPage(main, _params, config) {
})),
),
globalRulesCard(),
h(
'section',
{ class: 'card' },
h('h2', {}, 'FanFicFare settings (personal.ini)'),
card(
'FanFicFare settings (personal.ini)',
h(
'p',
{ class: 'hint' },
+93 -3
View File
@@ -451,6 +451,25 @@ label:has(> input[type='checkbox']:disabled) {
margin: 0;
}
ul.hint {
padding-left: 20px;
}
/* The fields of the download template: each name with its meaning beside it. */
.fields {
display: grid;
grid-template-columns: max-content 1fr;
gap: 2px 12px;
max-width: 80ch;
margin: 4px 0 8px;
color: var(--muted);
font-size: var(--text-sm);
dd {
margin: 0;
}
}
.notice {
margin: 12px 0;
padding: 10px 14px;
@@ -556,6 +575,11 @@ button.danger:hover:not(:disabled) {
background: var(--err-bg);
}
input:not([type='checkbox'], [type='radio']):disabled {
opacity: 0.55;
cursor: not-allowed;
}
button:disabled {
opacity: 0.55;
cursor: default;
@@ -699,6 +723,35 @@ textarea {
line-height: 1.45;
}
/* A user name over a password, with the labels in one column. */
.credentials {
display: grid;
grid-template-columns: max-content minmax(0, 1fr);
align-items: center;
gap: 8px 12px;
max-width: 48em;
margin-top: 8px;
input {
width: 18em;
}
input[readonly] {
background: var(--card-bg);
font-family: var(--mono);
cursor: default;
}
.controls {
margin-top: 0;
}
}
.template {
flex: 1 1 24em;
max-width: 48em;
}
.metadata input:not([type='checkbox']) {
width: 100%;
max-width: 32em;
@@ -822,18 +875,45 @@ textarea {
display: block;
}
/* The column headings become a row of sort chips. The chip of the sorted column is tinted and
shows the direction. The other chips show no arrow. */
.stories-table thead tr {
display: flex;
flex-wrap: wrap;
gap: 8px 16px;
margin-bottom: 8px;
gap: 6px;
margin-bottom: 10px;
}
.stories-table thead tr::before {
content: 'Sort by';
flex-basis: 100%;
color: var(--muted);
font-size: var(--text-sm);
}
.stories-table th {
padding: 4px 0;
padding: 0;
border: 0;
}
.stories-table th button {
inline-size: auto;
padding: 3px 10px;
border: 1px solid var(--border);
border-radius: 999px;
background: var(--surface);
font-size: var(--text-sm);
}
.stories-table th[aria-sort='none'] button::after {
display: none;
}
.stories-table th:not([aria-sort='none']) button {
background: var(--info-bg);
color: var(--info);
}
.stories-table tbody tr,
.chapters-table tbody tr {
border: 1px solid var(--border);
@@ -1039,6 +1119,16 @@ textarea {
margin-top: 12px;
}
/* Help that folds shut under a setting, such as the help for the download template. */
.help {
margin: 0 0 12px;
> summary {
color: var(--muted);
font-size: var(--text-sm);
}
}
/* Toasts */
#toasts {
+12
View File
@@ -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
+42 -21
View File
@@ -2,7 +2,6 @@
import base64
import hashlib
import io
import logging
import os
import re
@@ -13,14 +12,13 @@ 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
from bindery.ingest import sanitize_filename
from bindery.jobs import METADATA_FIELDS, MODE_LABELS, DownloadMode
from bindery.rules import RuleError
@@ -70,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,
@@ -149,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 +282,8 @@ def app_config() -> Any:
"update_interval": config.UPDATE_INTERVAL_HOURS,
"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,24 +432,12 @@ def delete_chapters(story_id: int) -> Any:
@app.get("/api/stories/<int:story_id>/epub")
def story_epub(story_id: int) -> Any:
file, title = orchestrator.build_for_download(story_id)
name = f"{sanitize_filename(title)}.epub"
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/<int:story_id>/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.
+1 -1
View File
@@ -9,7 +9,7 @@ test('a user who is not an admin gets no settings link and no settings page', as
t.after(stopPolling);
const backend = server(t);
backend.on('GET', 'auth/check', { authenticated: true, mode: 'proxy', user: 'bob', admin: false });
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false });
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false, download_fields: {} });
backend.on('GET', 'jobs', []);
backend.on('GET', 'stories', []);
await import('../../bindery/static/js/main.js');
+1 -1
View File
@@ -11,7 +11,7 @@ test('login retry, navigation, session expiry and logout preserve the authentica
const backend = server(t);
let auth = { authenticated: false, mode: 'cwa', user: null };
backend.on('GET', 'auth/check', () => auth);
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false });
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false, download_fields: {} });
backend.on('GET', 'jobs', []);
backend.on('GET', 'stories', []);
backend.on('GET', 'rules', { rules: [], paragraphs: false });
+48 -1
View File
@@ -16,6 +16,7 @@ const config = {
metadata_fields: { title: 'Title', authors: 'Authors', comments: 'Description', series: 'Series', pubdate: 'Published', tags: 'Tags', languages: 'Language' },
mode_labels: { new: 'New chapters' }, library_available: true, cwa_enabled: true, update_interval: 2,
job_list_max_hours: 720, job_list_max_count: 500,
download_fields: { title: 'The title', id: 'The story ID' },
};
const rule = { pattern: 'before', replacement: 'after', flags: 'i', note: '', enabled: true };
const emptyRules = { rules: [], paragraphs: null, global_paragraphs: true };
@@ -54,7 +55,7 @@ for (const value of ['7', '']) {
await settle();
api.on('PUT', 'settings', {});
field(main, 'recent-chapters').value = value;
await click(section(main, 'Settings'), 'Save');
await click(section(main, 'Recheck recent chapters'), 'Save');
assert.deepEqual(api.sent('PUT', 'settings')[0].body, { recent_chapters: value === '' ? 0 : 7 });
assert.match(ui.toasts.textContent, /Saved/);
});
@@ -108,6 +109,52 @@ test('the jobs list settings load and save all four fields', async () => {
assert.match(ui.toasts.textContent, /Saved/);
});
test('each group of settings is an open card, and the download help starts closed', async () => {
settingsPage(main, {}, config);
await settle();
const cards = main.children.filter((el) => el.tagName !== 'h1');
assert.deepEqual(cards.map((el) => [el.tagName, el.className, el.open, el.children[0].textContent]), [
['details', 'card', true, 'Recheck recent chapters'],
['details', 'card', true, 'Uploads to CWA'],
['details', 'card', true, 'Book Downloads'],
['details', 'card', true, 'Jobs list'],
['details', 'card', true, 'Global replacement rules'],
['details', 'card', true, 'FanFicFare settings (personal.ini)'],
]);
const help = nodes(section(main, 'Book Downloads')).filter((el) => el.className === 'help');
assert.deepEqual(help.map((el) => [el.tagName, el.open, el.children[0].textContent]), [
['details', undefined, 'File name template help'],
['details', undefined, 'Date format help'],
]);
});
test('the download template loads, lists its fields, and saves the stored form', async () => {
api.on('GET', 'settings', {
recent_chapters: 0, personal_ini: '', download_template: '{title}', download_date_format: '%b, %Y',
});
settingsPage(main, {}, config);
await settle();
const card = section(main, 'Book Downloads');
assert.equal(field(main, 'download-template').value, '{title}');
assert.equal(field(main, 'download-date-format').value, '%b, %Y');
assert.ok(find(card, 'code', '{id}'));
assert.match(card.textContent, /\{title\}The title\{id\}The story ID/);
field(main, 'download-template').value = ' {title} - {authors} ';
field(main, 'download-date-format').value = '';
api.on('PUT', 'settings', { download_template: '{title} - {authors}', download_date_format: '%b, %Y' });
await click(card, 'Save');
assert.deepEqual(api.sent('PUT', 'settings')[0].body, {
opds_enabled: false, download_template: ' {title} - {authors} ', download_date_format: '',
});
assert.equal(field(main, 'download-template').value, '{title} - {authors}');
assert.equal(field(main, 'download-date-format').value, '%b, %Y');
api.on('PUT', 'settings', { error: 'Bindery does not support template functions' }, 400);
field(main, 'download-template').value = '{title:uppercase()}';
await click(card, 'Save');
assert.equal(field(main, 'download-template').value, '{title:uppercase()}');
assert.match(ui.toasts.textContent, /does not support template functions/);
});
test('adding stories keeps input on rejection and clears it after a mixed result', async () => {
await mountStories({ ...config, library_available: false });
const urls = nodes(main).find((el) => el.tagName === 'textarea');
+1
View File
@@ -48,6 +48,7 @@ const settle = () => new Promise((resolve) => setImmediate(resolve));
const config = {
metadata_fields: { title: 'Title' }, mode_labels: {}, library_available: true, cwa_enabled: true, update_interval: 0,
job_list_max_hours: 720, job_list_max_count: 500,
download_fields: { title: 'The title', id: 'The story ID' },
};
const story = {
id: 1, title: 'Story', author: 'Author', url: 'https://example.com/story', site: 'example.com',
+24
View File
@@ -114,6 +114,30 @@ test('sortable headings show direction and remain usable above story cards on na
assert.doesNotMatch(css, /\.stories-table thead\s*\{[^}]*clip-path/);
});
test('on narrow screens the headings are sort chips, and only the sorted chip is tinted with an arrow', () => {
const start = css.indexOf('@media (max-width: 600px)', css.indexOf('.stories-table th button'));
const mobile = css.slice(start);
assert.equal(declarations('.stories-table thead tr', mobile)['flex-wrap'], 'wrap');
const caption = declarations('.stories-table thead tr::before', mobile);
assert.equal(caption.content, "'Sort by'");
assert.equal(caption['flex-basis'], '100%');
const chip = declarations('.stories-table th button', mobile);
assert.equal(chip['border-radius'], '999px');
assert.equal(chip['inline-size'], 'auto');
assert.equal(chip.padding, '3px 10px');
assert.equal(chip['font-size'], 'var(--text-sm)');
assert.equal(chip.background, 'var(--surface)');
assert.equal(declarations(".stories-table th[aria-sort='none'] button::after", mobile).display, 'none');
const sorted = declarations(".stories-table th:not([aria-sort='none']) button", mobile);
assert.equal(sorted.background, 'var(--info-bg)');
assert.equal(sorted.color, 'var(--info)');
assert.equal(sorted['border-color'], undefined, 'The sorted chip keeps the soft border');
// The wide layout keeps plain headings.
const wide = css.slice(0, start);
assert.doesNotMatch(wide, /\.stories-table thead tr::before/);
assert.equal(declarations('.stories-table th button', wide).border, '0');
});
test('sortable headings fill the column with one clickable highlight without shifting layout', () => {
assert.equal(declarations('.stories-table th').padding, '0');
const button = declarations('.stories-table th button');
+1 -1
View File
@@ -11,7 +11,7 @@ test('local login, retry, session expiry and logout use the existing form', asyn
const backend = server(t);
let auth = { authenticated: false, mode: 'local', user: null };
backend.on('GET', 'auth/check', () => auth);
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false });
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false, download_fields: {} });
backend.on('GET', 'stories', []);
backend.on('GET', 'jobs', []);
backend.on('GET', 'expired', { error: 'Session ended' }, 401);
+127
View File
@@ -0,0 +1,127 @@
import assert from 'node:assert/strict';
import { test } from 'node:test';
import { Element, change, click, field, find, install, nodes, section, server, settle } from './flow-helpers.mjs';
install();
const { settingsPage } = await import('../../bindery/static/js/settings.js');
async function open(t, settings, authMode = 'proxy') {
install();
t.mock.timers.enable({ apis: ['setTimeout'] });
const backend = server(t);
const stored = { download_template: '{title}', download_date_format: '%Y', personal_ini: '', ...settings };
backend.on('GET', 'settings', stored);
backend.on('GET', 'rules', { rules: [] });
backend.on('GET', 'stories', []);
backend.on('PUT', 'settings', ({ body }) => ({ ...stored, ...body }));
backend.on('PUT', 'settings/opds-password', { opds_user: 'alice', opds_password_set: true });
const main = new Element('main');
settingsPage(main, {}, { cwa_enabled: false, download_fields: {}, auth_mode: authMode });
await settle();
return { backend, main, card: section(main, 'Book Downloads') };
}
for (const enabled of [false, true]) {
test(`OPDS loads ${enabled}, appears first in Book Downloads, and saves both states`, async (t) => {
const { backend, main, card } = await open(t, { opds_enabled: enabled });
const toggle = field(card, 'opds-enabled');
assert.equal(toggle.checked, enabled);
assert.equal(card.children[1].tagName, 'label');
assert.equal(card.children[1].children[0], toggle);
assert.equal(card.children[1].textContent, ' Enable OPDS');
assert.equal(toggle.getAttribute('aria-describedby'), 'opds-note');
assert.match(field(card, 'opds-note').textContent, /off by default/);
assert.equal(find(field(card, 'opds-note'), 'a', 'the OPDS catalog').getAttribute('href'), 'opds');
assert.equal(nodes(main).filter((node) => node.getAttribute('id') === 'opds-enabled').length, 1);
for (const value of [!enabled, enabled]) {
change(toggle, value);
await click(card, 'Save');
assert.deepEqual(backend.sent('PUT', 'settings').at(-1).body, {
opds_enabled: value, download_template: '{title}', download_date_format: '%Y',
});
assert.equal(toggle.checked, value);
}
backend.on('PUT', 'settings', { opds_enabled: false, download_template: '{title}', download_date_format: '%Y' });
change(toggle, true);
await click(card, 'Save');
assert.equal(toggle.checked, false, 'The saved response sets the checkbox state');
});
}
const LOGIN = {
none: /Bindery asks for no login, so the catalog needs none/,
local: /sends your Bindery user name and password/,
cwa: /sends your CWA user name and password/,
proxy: /The proxy skips its login for the catalog.*Each user sets an OPDS password.*user name from the proxy/,
};
for (const [mode, text] of Object.entries(LOGIN)) {
test(`the note names the login of the ${mode} mode, and only the proxy mode shows the password`, async (t) => {
const { card } = await open(t, { opds_enabled: true, opds_password_set: false, opds_user: 'alice' }, mode);
assert.match(field(card, 'opds-note').textContent, text);
assert.doesNotMatch(field(card, 'opds-note').textContent, /BINDERY_API_KEY|Bearer|X-Api-Key/);
for (const id of ['opds-user', 'opds-password', 'opds-password-status']) {
assert.equal(field(card, id) !== undefined, mode === 'proxy', id);
}
assert.equal(find(card, 'button', 'Set password') !== undefined, mode === 'proxy');
});
}
test('the password field is grey until the box is checked, and follows the box at once', async (t) => {
const { card } = await open(t, { opds_enabled: false, opds_password_set: true, opds_user: 'alice' });
const password = field(card, 'opds-password');
const button = find(card, 'button', 'Reset password');
assert.equal(password.disabled, true);
assert.equal(button.disabled, true);
change(field(card, 'opds-enabled'), true);
assert.equal(password.disabled, false);
assert.equal(button.disabled, false);
change(field(card, 'opds-enabled'), false);
assert.equal(password.disabled, true);
assert.equal(button.disabled, true);
});
for (const isSet of [false, true]) {
test(`the OPDS password ${isSet ? 'is set' : 'is not set'}, and the button sets or resets it`, async (t) => {
const { backend, card } = await open(t, { opds_enabled: true, opds_password_set: isSet, opds_user: 'alice' });
const password = field(card, 'opds-password');
const status = field(card, 'opds-password-status');
const user = field(card, 'opds-user');
assert.equal(user.value, 'alice');
assert.equal(user.getAttribute('readonly'), '');
assert.equal(user.disabled, false, 'The user name stays selectable');
assert.equal(user.getAttribute('aria-describedby'), 'opds-password-status');
assert.equal(find(card, 'label', 'OPDS user name').getAttribute('for'), 'opds-user');
assert.equal(find(card, 'label', 'OPDS password').getAttribute('for'), 'opds-password');
assert.equal(password.disabled, false);
assert.equal(password.getAttribute('type'), 'password');
assert.equal(password.getAttribute('autocomplete'), 'new-password');
assert.equal(password.getAttribute('minlength'), '8');
assert.equal(password.value, '');
assert.match(status.textContent, /enter the user name and the password in your reader/i);
assert.match(status.textContent, isSet ? /You have an OPDS password/ : /You have no OPDS password yet/);
assert.ok(find(card, 'button', isSet ? 'Reset password' : 'Set password'));
password.value = 'long enough';
await click(card, isSet ? 'Reset password' : 'Set password');
assert.deepEqual(backend.sent('PUT', 'settings/opds-password')[0].body, { password: 'long enough' });
assert.equal(password.value, '', 'The field is empty after the save');
assert.match(status.textContent, /You have an OPDS password/);
assert.equal(password.disabled, false);
assert.ok(find(card, 'button', 'Reset password'));
// The Save button of the card never sends the password.
password.value = 'not sent';
await click(card, 'Save');
assert.equal(backend.sent('PUT', 'settings/opds-password').length, 1);
assert.equal('password' in backend.sent('PUT', 'settings')[0].body, false);
assert.equal(password.disabled, false, 'A save keeps the field in step with the box');
});
}
test('a rejected OPDS password keeps the field and the status', async (t) => {
const { backend, card } = await open(t, { opds_enabled: true, opds_password_set: false });
backend.on('PUT', 'settings/opds-password', { error: 'The OPDS password must be text of 8 to 256 characters' }, 400);
field(card, 'opds-password').value = 'short';
await click(card, 'Set password');
assert.equal(field(card, 'opds-password').value, 'short');
assert.match(field(card, 'opds-password-status').textContent, /You have no OPDS password yet/);
assert.ok(find(card, 'button', 'Set password'));
});
+1 -1
View File
@@ -15,7 +15,7 @@ test('a 401 while logged in through the proxy reloads the page', async (t) => {
});
const backend = server(t);
backend.on('GET', 'auth/check', { authenticated: true, mode: 'proxy', user: 'alice' });
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false });
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false, download_fields: {} });
backend.on('GET', 'jobs', []);
backend.on('GET', 'stories', []);
backend.on('GET', 'expired', { error: 'Session ended' }, 401);
+1 -1
View File
@@ -31,7 +31,7 @@ test('an open story has a tab with its title, also on its chapter pages', async
t.after(stopPolling);
const backend = server(t);
backend.on('GET', 'auth/check', { authenticated: true, mode: 'none', user: '' });
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false });
backend.on('GET', 'config', { metadata_fields: {}, mode_labels: {}, library_available: false, download_fields: {} });
backend.on('GET', 'jobs', []);
backend.on('GET', 'stories', []);
backend.on('GET', 'settings', { recent_chapters: 0, personal_ini: '' });
+1 -1
View File
@@ -7,7 +7,7 @@ const { settingsPage } = await import('../../bindery/static/js/settings.js');
const { storyPage } = await import('../../bindery/static/js/story.js');
const { storiesPage } = await import('../../bindery/static/js/stories.js');
const config = { cwa_enabled: true, job_list_max_hours: 720, job_list_max_count: 500 };
const config = { cwa_enabled: true, job_list_max_hours: 720, job_list_max_count: 500, download_fields: {} };
const book = { id: 7, uuid: 'book-7', title: 'Library Title', authors: ['Writer'] };
const details = (story, extra = {}) => ({
story: { id: 1, title: 'Story', author: 'Writer', url: 'https://example.com/story', site: 'example.com',
+5
View File
@@ -472,6 +472,11 @@ 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,
"story_jobs_count": 10,
"main_jobs_hours": 6,
+75 -1
View File
@@ -142,11 +142,85 @@ def test_download_builds_from_stored_chapters_without_changing_them(client, stor
response = client.get(f"/api/stories/{sid}/epub")
assert response.status_code == 200
assert response.data.startswith(b"PK")
assert "filename=A_B.epub" in response.headers["Content-Disposition"]
assert 'filename="A_B - Writer.epub"' in response.headers["Content-Disposition"]
assert o.store.revision_html(cid, 1) == original
response.close()
def download_name(client, sid):
response = client.get(f"/api/stories/{sid}/epub")
response.close()
return response.headers["Content-Disposition"]
def test_download_template_names_the_file(client, story):
sid, _ = story
site = {"title": "The Book", "authors": ["Ann Writer"], "series": "The Saga", "series_index": 2.0}
o.store.update_story(sid, metahtml=metahtml(), title="The Book", site_values=site)
assert client.get("/api/config").json["download_fields"]["id"].startswith("The CWA book ID")
saved = client.put("/api/settings", json={"download_template": " {series:|| - }{title} - {id} "}, headers=H)
assert saved.json["download_template"] == "{series:|| - }{title} - {id}"
assert f'filename="The Saga - The Book - {sid}.epub"' in download_name(client, sid)
# The default moves the article to the end, as Calibre's default does.
client.put("/api/settings", json={"download_template": ""}, headers=H)
assert 'filename="Book, The - Ann Writer.epub"' in download_name(client, sid)
client.put("/api/settings", json={"download_template": "{series_sort} - {title} - {id}"}, headers=H)
assert f'filename="Saga, The - The Book - {sid}.epub"' in download_name(client, sid)
# The ID is that of the CWA book, when the story has one.
o.store.update_story(sid, calibre_id=77, pinned={"title": "Caf\u00e9", "series": ["", None]})
client.put("/api/settings", json={"download_template": "{series:|| - }{title} - {id}"}, headers=H)
name = download_name(client, sid)
assert 'filename="Cafe - 77.epub"' in name and "filename*=UTF-8''Caf%C3%A9%20-%2077.epub" in name
# A template with folders gives the name of the file alone.
client.put("/api/settings", json={"download_template": "{author_sort}/{title}"}, headers=H)
assert "filename*=UTF-8''Caf%C3%A9.epub" in download_name(client, sid)
# An empty template restores the default.
saved = client.put("/api/settings", json={"download_template": " "}, headers=H).json
assert saved["download_template"] == "{title_sort} - {authors}"
assert "filename*=UTF-8''Caf%C3%A9%20-%20Ann%20Writer.epub" in download_name(client, sid)
def test_date_format_of_download_names(client, story):
sid, _ = story
o.store.update_story(sid, metahtml=metahtml(), site_values={"pubdate": "2020-05-06T00:00:00"})
client.put("/api/settings", json={"download_template": "{title} {pubdate}"}, headers=H)
assert 'filename="Story May, 2020.epub"' in download_name(client, sid)
saved = client.put("/api/settings", json={"download_date_format": " %Y-%m-%d "}, headers=H).json
assert saved["download_date_format"] == "%Y-%m-%d"
assert "filename=Story 2020-05-06.epub" in download_name(client, sid).replace('"', "")
response = client.put("/api/settings", json={"download_date_format": 5}, headers=H)
assert (response.status_code, response.json) == (400, {"error": "The date format must be text"})
response = client.put("/api/settings", json={"download_date_format": "%Y" * 51}, headers=H)
assert (response.status_code, response.json) == (400, {"error": "A date format can have 100 characters at most"})
# An empty format restores the default.
assert client.put("/api/settings", json={"download_date_format": ""}, headers=H).json["download_date_format"] == (
"%b, %Y"
)
def test_download_template_that_fails_gives_the_title(client, story, caplog):
sid, _ = story
o.store.update_story(sid, metahtml=metahtml())
for template in ("{title:d}", "{series}/{series_index}"):
client.put("/api/settings", json={"download_template": template}, headers=H)
assert "filename=Story.epub" in download_name(client, sid)
assert "The download template fails for https://www.royalroad.com/fiction/1234" in caplog.text
assert caplog.text.count("The download template fails") == 1
def test_bad_download_template_is_not_saved(client):
client.put("/api/settings", json={"download_template": "{title}"}, headers=H)
for template, error in [
("{title:uppercase()}", "Bindery does not support template functions, such as {title:uppercase()}"),
("{title", "The braces of the template do not match: expected '}' before end of string"),
(7, "The download template must be text"),
]:
response = client.put("/api/settings", json={"download_template": template, "recent_chapters": 9}, headers=H)
assert (response.status_code, response.json) == (400, {"error": error})
assert client.get("/api/settings").json["download_template"] == "{title}"
assert client.get("/api/settings").json["recent_chapters"] == 0
def test_unexpected_backend_error_returns_json(client, monkeypatch, caplog):
monkeypatch.setattr(o, "list_stories", Mock(side_effect=RuntimeError("broken")))
response = client.get("/api/stories")
+228
View File
@@ -0,0 +1,228 @@
"""The file name templates, against the results of Calibre 9.2.1.
Each expected text is what Calibre gives for the same book and template.
"""
import pytest
from bindery import filename_template as ft
BOOK = {
"title": "The Book",
"authors": ["Ann Writer", "Ben Reader"],
"series": "The Saga",
"series_index": 2.345,
"languages": ["eng", "fra"],
"tags": ["Space", "Fiction"],
"pubdate": "2020-05-06T12:00:00+00:00",
"publisher": "site.example",
}
NO_SERIES = {**BOOK, "series": "", "series_index": None}
def expand(template, order="library_order", **changes):
return ft.expand(template, ft.prepare({**BOOK, **changes}, 42, order))
# name, template, text from Calibre, series index
REFERENCE = [
("title", "{title}", "Book, The", 2.345),
("series", "{series}", "Saga, The", 2.345),
("fraction", "{series_index}", "2.35", 2.345),
("languages", "{languages}", "eng,fra", 2.345),
("tags", "{tags}", "Fiction, Space", 2.345),
("send_date", "{pubdate}", "May, 2020", 2.345),
("space_padding", "x{series_index:>5s}x", "x 2x", 2),
("zero_padding", "{series_index:0>3s}", "002", 2),
("numeric_format", "{series_index:03d}", "002", 2),
("first_character", "{author_sort[0]}", "W", 2),
("conditional", "{series:|| - }{title}", "Saga, The - Book, The", 2),
("case_insensitive", "{TITLE}", "Book, The", 2),
("unknown", "x{not_a_field}x", "xx", 2),
("literal_braces", "{{title}}", "{title}", 2),
]
@pytest.mark.parametrize("name,template,expected,index", REFERENCE, ids=[case[0] for case in REFERENCE])
def test_reference_cases(name, template, expected, index):
assert expand(template, series_index=index) == expected
def test_reference_cases_for_a_missing_series_and_for_alphabetic_order():
values = ft.prepare(NO_SERIES, 42)
assert ft.expand("{series}{series_index:| - | - }{title}", values) == "Book, The"
assert expand("{title}", "strictly_alphabetic") == "The Book"
assert expand("{series}", "strictly_alphabetic") == "The Saga"
def test_series_sort_is_a_bindery_field():
assert expand("{series_sort}") == expand("{series_sort}", "strictly_alphabetic") == "Saga, The"
assert ft.expand("x{series_sort}x", ft.prepare(NO_SERIES, 42)) == "xx"
@pytest.mark.parametrize(
"template,expected",
[
("{authors}", "Ann Writer & Ben Reader"),
("{author}", "Ann Writer & Ben Reader"),
("{author_sort}", "Writer, Ann & Reader, Ben"),
("{title_sort}", "Book, The"),
("{publisher} {id}", "site.example 42"),
("{title:.3s|<|>}", "<Boo>"),
("{title:*^15}", "***Book, The***"),
("{id:05d} {id:x} {id:.2f}", "00042 2a 42.00"),
("{title[0]}{title[1]}", "Bo"),
("{title} {{{id}}}", "Book, The {42}"),
(" {title} -\t{authors} ", "Book, The - Ann Writer & Ben Reader"),
("{ title }{other}", ""),
("a|b|c{title:|-|}", "a|b|c-Book, The"),
("{series}/{series_index:0>2s| | - }{title}", "Saga, The/ 2.35 - Book, The"),
],
)
def test_more_templates_give_the_text_of_calibre(template, expected):
assert expand(template) == expected
def test_values_follow_the_rules_of_calibre():
assert expand("{title} - {authors}", title="A/B: C\\D", authors=["X/Y"]) == "A_B: C_D - X_Y"
assert expand("{authors}", authors=["Rock & Roll", "", "Solo"]) == "Rock && Roll & Solo"
assert expand("{tags}", tags=["/root", "alpha", "Beta", "1st", "_x"]) == "_x, _root, 1st, alpha, Beta"
assert expand("{tags}", tags=["F/M", "Hurt-Comfort", "hurt comfort", "Hurt", "x2", "x10", "é", "e", "f"]) == (
"e, é, f, F_M, Hurt, hurt comfort, Hurt-Comfort, x10, x2"
)
assert expand("{tags}|{languages}|{pubdate}", tags=[], languages="", pubdate="") == "||"
assert expand("{languages}", languages="en") == "en"
assert expand("{pubdate} {timestamp}", pubdate="1999-12-31T23:59:59", timestamp="2026-10-03") == (
"Dec, 1999 Oct, 2026"
)
assert expand("{pubdate}", pubdate="not a date") == ""
def test_date_format():
# Calibre's helper_date reference case.
assert ft.expand("{pubdate}", ft.prepare(BOOK, 42, date_format="%b %Y")) == "May 2020"
assert ft.expand("{pubdate}", ft.prepare(BOOK, 42, date_format="%d/%m/%Y")) == "06_05_2020"
assert ft.expand("x{pubdate}x", ft.prepare(BOOK, 42, date_format="")) == "xx"
ft.check_date_format("%Y-%m-%d " * 10)
with pytest.raises(ft.TemplateError, match="A date format can have 100 characters at most"):
ft.check_date_format("x" * 101)
# A series without a number has no series_index, and a book without a series has neither.
assert expand("{series}{series_index:| [|]}", series_index=None) == "Saga, The"
assert ft.expand("{series_index:03d}|{series}", ft.prepare(NO_SERIES, 42)) == "|"
@pytest.mark.parametrize(
"title,expected",
[
("A Tale", "Tale, A"),
("An Apple", "Apple, An"),
("the lower", "lower, the"),
("Theory", "Theory"),
("The", "The"),
(" The Spaced ", "Spaced, The"),
("'The Quoted'", "Quoted, The"),
("“The Curly”", "Curly, The"),
("„The Low“", "Low, The"),
("The 'Inner", "Inner, The"),
('""', ""),
("Le Monde", "Le Monde"),
],
)
def test_title_sort(title, expected):
assert ft.title_sort(title) == expected
@pytest.mark.parametrize(
"author,expected",
[
("Ann Writer", "Writer, Ann"),
("Dr. John Smith Jr.", "Smith, John Jr."),
("John Smith MD PhD", "Smith, John MD PhD"),
("Mr Smith", "Smith"),
("Dr Mr", "Dr Mr"),
("Jr Sr", "Jr Sr"),
("xX_Dark_Xx", "xX_Dark_Xx"),
("Smith, John", "Smith, John"),
("John (Jack) Doe", "Doe, John"),
("A (b c", "A (b c"),
("Acme Media", "Acme Media"),
("Ludwig van Beethoven", "Beethoven, Ludwig van"),
("J. R. R. Tolkien", "Tolkien, J. R. R."),
],
)
def test_author_sort(author, expected):
assert ft.author_sort(author) == expected
@pytest.mark.parametrize(
"index,expected",
[(1, "1"), (2.0, "2"), (0, "0"), (2.5, "2.5"), (1.10, "1.1"), (3.14159, "3.14"), (2.001, "2."), (10.999, "11.")],
)
def test_series_index(index, expected):
assert ft.series_index(index) == expected
@pytest.mark.parametrize(
"template,error",
[
("{title:uppercase()}", "does not support template functions"),
("{series:ifempty(none)}", "does not support template functions"),
("{title:'uppercase($)'}", "does not support template functions"),
("{title:.3s:'uppercase($)'}", "does not support template functions"),
("program: field('title')", "does not support the program modes"),
("python: def evaluate(book, context): return ''", "does not support the program modes"),
("{title!r}", "does not support the field"),
("{title.upper}", "does not support the field"),
("{title[x]}", "does not support the field"),
("{title:{id}}", "does not support the field"),
("{}", "A field needs a name"),
("{0}", "A field needs a name"),
("{title", "The braces of the template do not match"),
("title}", "The braces of the template do not match"),
("{title:5q}", "The format '5q' cannot take"),
("{series:|a|b|c}", "The format '|a' cannot take"),
("{id:.3d}", "The format '.3d' cannot take"),
("{title:>256s}", "A format can ask for 255 characters at most"),
("{id:.999f}", "A format can ask for 255 characters at most"),
("x" * 1025, "A template can have 1024 characters at most"),
],
)
def test_check_refuses_what_bindery_does_not_support(template, error):
with pytest.raises(ft.TemplateError, match=error.replace("|", "\\|")):
ft.check(template)
def test_check_accepts_the_templates_of_calibre():
for template in ("", "{author_sort}/{title}/{title} - {authors}", "{title:>255s}", "x" * 1024):
ft.check(template)
@pytest.mark.parametrize("template", ["{series_index:03d}", "{title:d}", "{title[50]}", "{id:c}{title:c}"])
def test_a_template_can_fail_for_one_book(template):
ft.check(template)
with pytest.raises(ft.TemplateError):
expand(template)
def file_name(template, **changes):
return ft.file_name(template, ft.prepare({**BOOK, **changes}, 42))
def test_file_name_is_the_last_part_of_the_path():
assert file_name("{author_sort}/{title}/{title} - {authors}") == "Book, The - Ann Writer & Ben Reader"
assert file_name("{series}//{title} /") == "Book, The"
assert file_name("{author_sort[0]}/{series}", series="") == "W"
assert file_name("{series}/{other}", series="") == ""
assert file_name(" / ") == ""
def test_file_name_is_safe():
assert file_name("{title}?", title="A/B: C\\D") == "A_B_ C_D_"
assert file_name("<{id:c}>") == "___"
assert file_name("{title}", title="con") == "_con"
assert file_name("{title}.{id}", title="NUL") == "_NUL.42"
assert file_name("{title}", title="Console") == "Console"
assert file_name("{title}...") == "Book, The"
long = file_name("{title}", title="é" * 150)
assert long == "é" * 100
assert len(file_name("{title}", title="\U0001f600" * 150).encode()) == 200
+267
View File
@@ -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}"}
+329
View File
@@ -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
+186
View File
@@ -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"
+9
View File
@@ -137,6 +137,15 @@ def test_cwa_off_shows_in_the_config_and_refuses_cwa_logins(client, monkeypatch,
assert response.json["error"].startswith("AUTH_METHOD=cwa needs CWA, but CWA_ENABLED is false.")
def test_without_cwa_the_download_template_uses_the_story_id(client, cwa_off):
assert client.get("/api/config").json["download_fields"]["id"] == "The Bindery story ID"
story_id = add_story(calibre_id=77)
client.put("/api/settings", json={"download_template": "{title} {id}"}, headers=H)
response = client.get(f"/api/stories/{story_id}/epub")
response.close()
assert f'filename="My Title {story_id}.epub"' in response.headers["Content-Disposition"]
def test_link_setting_defaults_on_and_persists(client, monkeypatch, tmp_path, library):
assert client.get("/api/settings").json["link_books"] is True
response = client.put("/api/settings", json={"link_books": False}, headers=H)