From 5f682d209b429abb62bf384b90b6a9c72f7fdcc5 Mon Sep 17 00:00:00 2001 From: Timothy Allen Date: Fri, 2 Oct 2026 23:52:22 +0200 Subject: [PATCH] Remove small duplicates in the settings, the logins and the names One function reads a true or false value. The download name uses the file name cleaning of the ingest folder. The login code reads the login mode from the configuration, and has one lockout message. The title helper of FanFicFare is now unescape_title, because cwa.py has another plain_title. --- bindery/auth.py | 17 +++++++---------- bindery/config.py | 18 ++++++++++++++---- bindery/fff.py | 2 +- bindery/orchestrator.py | 4 ++-- bindery/web.py | 19 ++++++++++--------- scripts/import_unlinked_epub_chapters.py | 4 ++-- tests/test_build.py | 2 +- tests/test_config.py | 5 ++++- 8 files changed, 41 insertions(+), 30 deletions(-) diff --git a/bindery/auth.py b/bindery/auth.py index 247baaa..2dfdfda 100644 --- a/bindery/auth.py +++ b/bindery/auth.py @@ -28,6 +28,7 @@ AUTH_NONE, AUTH_LOCAL, AUTH_CWA, AUTH_PROXY = "none", "local", "cwa", "proxy" MAX_FAILURES = 5 # for one user name from one address MAX_ADDRESS_FAILURES = 20 # for all user names from one address LOCKOUT_SECONDS = 15 * 60 +LOCKED = "Too many wrong passwords. Try again in 15 minutes." # Name endings that only a private network can resolve. LOCAL_SUFFIXES = (".local", ".lan", ".home", ".home.arpa", ".internal", ".localhost", ".localdomain") @@ -134,10 +135,6 @@ def config_error() -> str | None: return None -def active_mode() -> str: - return config.AUTH_METHOD - - def _allowed(name: str) -> bool: names = {n.casefold() for n in config.CWA_ALLOWED_USERS} return not names or name.casefold() in names @@ -149,7 +146,7 @@ def is_admin(identity: Identity) -> bool: Only the cwa and proxy modes have more than one user. The API key is always an admin. """ names = {n.casefold() for n in config.ADMIN_USERS} - if not names or identity.via in ("api_key", "none") or active_mode() == AUTH_LOCAL: + if not names or identity.via in ("api_key", "none") or config.AUTH_METHOD == AUTH_LOCAL: return True return identity.name.casefold() in names @@ -175,7 +172,7 @@ def current_identity( identity = _api_key_identity(headers) if identity: return identity - mode = active_mode() + mode = config.AUTH_METHOD if mode == AUTH_NONE: return Identity("", "none") if mode == AUTH_PROXY: @@ -256,12 +253,12 @@ def _login_ok(username: str, password: str, client: CwaClient) -> bool: def login_local(username: str, password: str, address: str) -> Identity: username = username.strip() - if active_mode() != AUTH_LOCAL: + if config.AUTH_METHOD != AUTH_LOCAL: raise AuthError("Bindery does not use local logins") if error := config_error(): raise AuthError(error) if limiter.locked(username, address): - raise AuthError("Too many wrong passwords. Try again in 15 minutes.") + raise AuthError(LOCKED) # Check the password even for a wrong user name, so both failures take the same work. password_ok = check_password_hash(config.LOCAL_PASSWORD_HASH, password) if not (password_ok and password and hmac.compare_digest(username.encode(), config.LOCAL_USERNAME.encode())): @@ -274,10 +271,10 @@ def login_local(username: str, password: str, address: str) -> Identity: def login_cwa(username: str, password: str, client: CwaClient, address: str) -> Identity: """Check a CWA login. Raises AuthError with a message for the user.""" username = username.strip() - if active_mode() != AUTH_CWA: + if config.AUTH_METHOD != AUTH_CWA: raise AuthError("Bindery does not use CWA logins") if limiter.locked(username, address): - raise AuthError("Too many wrong passwords. Try again in 15 minutes.") + raise AuthError(LOCKED) try: ok = ( _allowed(username) diff --git a/bindery/config.py b/bindery/config.py index 2e2e95e..3800571 100644 --- a/bindery/config.py +++ b/bindery/config.py @@ -8,8 +8,18 @@ from pathlib import Path logger = logging.getLogger(__name__) -def _bool(name: str, default: str = "false") -> bool: - return os.getenv(name, default).strip().lower() in ("true", "yes", "1", "y", "on") +def flag(value: str, default: bool = False) -> bool: + """A text value as true or false. A value that is neither gives the default.""" + value = value.strip().lower() + if value in ("true", "yes", "1", "y", "on"): + return True + if value in ("false", "no", "0", "n", "off"): + return False + return default + + +def _bool(name: str, default: bool = False) -> bool: + return flag(os.getenv(name, ""), default) def _list(name: str, default: str = "") -> list[str]: @@ -66,7 +76,7 @@ CWA_DB_PATH = _optional_path("CWA_DB_PATH") # Calibre-Web app.db, for logins CALIBRE_DB_PATH = _optional_path("CALIBRE_DB_PATH") # calibre metadata.db, for book IDs # false if the user has no CWA. Bindery then ignores the CWA settings above, and shows nothing about CWA. INGEST_DIR # still gets each built book, as a plain epub. Only a clear "false" turns CWA off. -CWA_ENABLED = os.getenv("CWA_ENABLED", "").strip().lower() not in ("false", "no", "0", "n", "off") +CWA_ENABLED = _bool("CWA_ENABLED", True) CWA_SETTINGS = ("CWA_URL", "CWA_USERNAME", "CWA_PASSWORD", "CWA_DB_PATH", "CALIBRE_DB_PATH") IGNORED_SETTINGS = [] if CWA_ENABLED else [name for name in CWA_SETTINGS if os.getenv(name, "").strip()] if not CWA_ENABLED: @@ -99,7 +109,7 @@ SESSION_COOKIE_SECURE = _bool("SESSION_COOKIE_SECURE") TRUST_PROXY = _bool("TRUST_PROXY") # Tests turn off the job threads. -START_THREADS = _bool("BINDERY_START_THREADS", "true") +START_THREADS = _bool("BINDERY_START_THREADS", True) SITE_DELAY = 5 # seconds between two jobs that go to a story site JOB_LIST_MAX_HOURS = 24 * 30 # the highest allowed value of a "jobs from the last N hours" setting diff --git a/bindery/fff.py b/bindery/fff.py index 5255c55..a94eb95 100644 --- a/bindery/fff.py +++ b/bindery/fff.py @@ -90,7 +90,7 @@ def escape_title(text: str) -> str: return html.escape(text, quote=False) -def plain_title(title: str) -> str: +def unescape_title(title: str) -> str: """A stored chapter title as text, for a message.""" return html.unescape(title) diff --git a/bindery/orchestrator.py b/bindery/orchestrator.py index da938b5..30da9a3 100644 --- a/bindery/orchestrator.py +++ b/bindery/orchestrator.py @@ -1192,7 +1192,7 @@ def _chapters_to_fetch( if mode in (DownloadMode.METADATA, DownloadMode.RECHECK): new_urls = [] selected = [c for c in chapters if c["id"] in job.chapter_ids] if mode == DownloadMode.RECHECK else [] - skipped = [fff.plain_title(c["title"]) for c in selected if not c["on_site"]] + skipped = [fff.unescape_title(c["title"]) for c in selected if not c["on_site"]] if mode in (DownloadMode.RECHECK_ALL, DownloadMode.FULL): refetch = [c for c in chapters if c["on_site"]] else: @@ -1254,7 +1254,7 @@ def _run_site_job(job: Job) -> str: def fetch(url: str, title: str, action: str) -> str | None: """Download a chapter and store its images. A chapter that fails goes in `failed`, and gives None.""" nonlocal done - name = fff.plain_title(title) + name = fff.unescape_title(title) _progress(job, f"{action} {done + 1} of {total}: {name}", done / max(total, 1)) done += 1 try: diff --git a/bindery/web.py b/bindery/web.py index 289e8d2..f7ce899 100644 --- a/bindery/web.py +++ b/bindery/web.py @@ -20,14 +20,15 @@ from bindery import auth, config, 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 logging.basicConfig(level=config.LOG_LEVEL, format="%(asctime)s %(levelname)s %(name)s: %(message)s") logger = logging.getLogger(__name__) -# API paths that need no login. -PUBLIC_PATHS = {"/api/health", "/api/auth/check", "/api/auth/login", "/api/auth/logout"} +# API paths that need no login. The login gate also lets /api/health through, before it reads a setting. +PUBLIC_PATHS = {"/api/auth/check", "/api/auth/login", "/api/auth/logout"} # The browser sends this header with every change. Another site cannot send it # without a CORS preflight, which Bindery never allows, so it stops cross-site requests. CSRF_HEADER = "X-Requested-With" @@ -219,7 +220,7 @@ def health() -> Any: @app.get("/api/auth/check") def auth_check() -> Any: identity = _identity() - mode = auth.active_mode() + mode = config.AUTH_METHOD return { "authenticated": identity is not None, "mode": mode, @@ -236,7 +237,7 @@ def auth_login() -> Any: username = str(data.get("username", "")) password = str(data.get("password", "")) address = auth.client_address(request.headers, request.remote_addr) - if auth.active_mode() == auth.AUTH_LOCAL: + if config.AUTH_METHOD == auth.AUTH_LOCAL: identity = auth.login_local(username, password, address) else: identity = auth.login_cwa(username, password, orchestrator.cwa(), address) @@ -244,8 +245,8 @@ def auth_login() -> Any: return {"error": str(e)}, 401 session.clear() session["user"] = identity.name - session["mode"] = auth.active_mode() - if auth.active_mode() == auth.AUTH_LOCAL: + session["mode"] = config.AUTH_METHOD + if config.AUTH_METHOD == auth.AUTH_LOCAL: session["local_password_version"] = auth.local_password_version() session.permanent = bool(data.get("remember")) logger.info(f"{identity.name} logged in") @@ -321,7 +322,7 @@ def import_epub(story_id: int | None = None) -> Any: raise ValueError("Choose an epub file") path = config.UPLOAD_DIR / f"{uuid.uuid4().hex}.epub" upload.save(path) - check = request.form.get("check", "true").lower() not in ("0", "false", "no", "off") + check = config.flag(request.form.get("check", ""), True) return orchestrator.queue_epub_file(path, os.path.basename(upload.filename), story_id, check) @@ -399,8 +400,8 @@ def delete_chapters(story_id: int) -> Any: @app.get("/api/stories//epub") def story_epub(story_id: int) -> Any: file, title = orchestrator.build_for_download(story_id) - name = re.sub(r'[\\/:*?"<>|]', "_", title.strip())[:200] or "story" - response = send_file(file, download_name=f"{name}.epub", as_attachment=True, mimetype="application/epub+zip") + 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 diff --git a/scripts/import_unlinked_epub_chapters.py b/scripts/import_unlinked_epub_chapters.py index 82ef985..02d533c 100644 --- a/scripts/import_unlinked_epub_chapters.py +++ b/scripts/import_unlinked_epub_chapters.py @@ -304,7 +304,7 @@ def remove_added_chapters(db_store: Store, story: dict, existing: list[dict]) -> return print(f"{story['title']} has {_count(len(added), 'chapter')} that this script added:") for c in added: - print(f" - {fff.plain_title(c['title'])}") + print(f" - {fff.unescape_title(c['title'])}") if not _confirm("Delete them and all their revisions?"): return conn = sqlite3.connect(config.DB_PATH) @@ -408,7 +408,7 @@ def main(argv: list[str] | None = None) -> None: raise ScriptError(f"{story['title']} has no stored chapters yet. Download it in Bindery first.") epub_chapters = read_epub(args.epub) - titles = [fff.plain_title(c["title"]) for c in existing] + titles = [fff.unescape_title(c["title"]) for c in existing] labels = [t + ("" if c["on_site"] else " (not on site)") for t, c in zip(titles, existing, strict=True)] epub_titles = [c["title"] for c in epub_chapters] diff --git a/tests/test_build.py b/tests/test_build.py index 31d953e..6291390 100644 --- a/tests/test_build.py +++ b/tests/test_build.py @@ -76,7 +76,7 @@ def test_a_title_with_entities_builds_and_imports_in_fanficfare_form(tmp_path): ElementTree.fromstring(z.read(name)) # An import stores the same title as a download, so the next update finds no rename. assert fff.read_epub(out).chapters[0]["title"] == title - assert fff.plain_title(title) == "Folly & Magic <1>" + assert fff.unescape_title(title) == "Folly & Magic <1>" assert fff.escape_title("Folly & Magic <1>") == title diff --git a/tests/test_config.py b/tests/test_config.py index 461c1ef..098cf57 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -24,7 +24,10 @@ def test_bool(monkeypatch): monkeypatch.setenv("X_TEST", value) assert not config._bool("X_TEST") monkeypatch.delenv("X_TEST") - assert config._bool("X_TEST", "true") + assert config._bool("X_TEST", True) + # A value that is not clear gives the default. + monkeypatch.setenv("X_TEST", "maybe") + assert config._bool("X_TEST", True) and not config._bool("X_TEST") def test_list(monkeypatch):