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):