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.
This commit is contained in:
+7
-10
@@ -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)
|
||||
|
||||
+14
-4
@@ -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
|
||||
|
||||
+1
-1
@@ -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)
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
+10
-9
@@ -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/<int:story_id>/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
|
||||
|
||||
|
||||
@@ -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]
|
||||
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user