from __future__ import annotations import base64 import collections import contextlib import hashlib import html import io import json import logging import math import mimetypes import os import re import secrets import shutil import sqlite3 import sys import threading import time from contextlib import asynccontextmanager, contextmanager from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Callable, Generator, Iterable, Literal, Sequence, cast, overload from urllib.parse import parse_qs, parse_qsl, quote, quote_plus, unquote, urlencode, urljoin, urlparse, urlsplit, urlunparse, urlunsplit import feedparser import httpx def _parse_month_first_pubdate(date_string: str): # Handles non-RFC-2822 dates like "May 11, 2026 19:15:50 +0000" (no day-of-week). try: dt = datetime.strptime(date_string.strip(), "%B %d, %Y %H:%M:%S %z") return dt.utctimetuple() except ValueError: return None feedparser.registerDateHandler(_parse_month_first_pubdate) # The import block below was never isort-ordered; this suppresses I001 for the whole block. from fastapi import FastAPI, Form, Query, Request # noqa: I001 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response, StreamingResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from PIL import Image as _PILImage from readability import Document from starlette.concurrency import run_in_threadpool from starlette.middleware.base import BaseHTTPMiddleware from starlette.middleware.gzip import GZipMiddleware from starlette.middleware.sessions import SessionMiddleware from services import entry_content, escalation_reprobe from services import bluesky from services import flaresolverr as flaresolverr_service from services import page_fetch from services import full_content_fetch, itad_waitlist from services import page_topics from services.errors import public_error from services.redirects import local_url from services import site_content_plugins from services import publish_date as publish_date_service from services import deviantart as deviantart_service from services import devto as devto_service from services import youtube_oauth as youtube_oauth_service from services import pinterest_oauth as pinterest_oauth_service from services import quire as quire_service from services import reddit as reddit_service from services import inoreader as inoreader_service from services import passwords from services import podcast_audio from services import podcast_feed_discovery from services import content_edits from services import link_canonical from services import saved_articles as saved_articles_service from services import saved_autofile as saved_autofile_service # noqa: F401 -- re-exported: routes/saved.py does `from main import saved_autofile_service` from services import archive_old_stars as archive_old_stars_service # noqa: F401 -- re-exported, same as above from services import refetch_batch from services import unstar_tagged as unstar_tagged_service from services import scraper_service from services import html_sanitize from services import tenancy from services import youtube_embeds from services import url_guard from services.webhooks import WEBHOOK_VALID_FORMATS from services.users import UserStore from services.email import send_article_email, send_digest_email from services import feed_discovery from services.feed_discovery import discover_feed_urls_ex # noqa: F401 -- re-exported: routes/feeds.py does `from main import discover_feed_urls_ex` from services.feed_refresh import FeedRefreshService from services.lead_images import LeadImageService, upgrade_image_size_param from services.reader_api import ReaderApi from services.starred_archive import StarredArchiveService from services.youtube import YouTubeDurationService from services.websub import WebSubService from services.fever import FeverService from services.greader import GReaderService from services.miniflux import MinifluxService from services.youtube_sync import sync_youtube_folder from state import ( _AUTOFETCH_HOST_COOLDOWN_S, _BROWSER_UA_CACHE_TTL, _DA_SYNC_MAX_AUTO_RESUMES, _FLARESOLVERR_FEEDS_CACHE_TTL, _INSTANCE_SETTING_TTL_SECONDS, _LAST_SEEN_THROTTLE_SECONDS, _MEDIA_SCAN_TTL_ERROR, _PROXY_DOWN_COOLDOWN_SECONDS, _PROXY_FEEDS_CACHE_TTL, _TAILSCALE_DOWN_COOLDOWN_SECONDS, _TAILSCALE_FEEDS_CACHE_TTL, _THUMB_FETCH_FAIL_CACHE, # noqa: F401 -- re-exported: routes/system.py's /thumb proxy does `main._THUMB_FETCH_FAIL_CACHE` _THUMB_FETCH_FAIL_LOCK, # noqa: F401 -- re-exported, same as above _THUMB_FETCH_FAIL_TTL, # noqa: F401 -- re-exported, same as above _app_settings_cache, _app_settings_cache_lock, _autofetch_failed_hosts, _autofetch_hosts_lock, _browser_ua_cache, _browser_ua_cache_at, _browser_ua_cache_lock, _da_sync_active, _da_sync_lock, _enhancement_inflight_feeds, _enhancement_inflight_lock, _feed_last_post_cache, _feed_last_post_cache_lock, _flaresolverr_feeds_cache, _flaresolverr_feeds_cache_at, _flaresolverr_feeds_cache_lock, _home_request_semaphore, # noqa: F401 -- re-exported: routes/home.py's `home` route does `main._home_request_semaphore` _instance_setting_cache, _instance_setting_cache_lock, _last_seen_touch, _last_seen_touch_lock, _login_failures, # noqa: F401 -- re-exported: routes/system.py's login rate-limit helpers do `main._login_failures` _login_failures_lock, # noqa: F401 -- re-exported, same as above _media_scan_in_progress, _media_scan_lock, _prefetch_header_log_lock, _prefetch_header_log_remaining, _proxy_down_lock, _proxy_down_until, _proxy_feeds_cache, _proxy_feeds_cache_at, _proxy_feeds_cache_lock, _scheduler_state, _scheduler_state_lock, _tag_alias_cache, _tag_alias_lock, _tailscale_down_lock, _tailscale_down_until, _tailscale_feeds_cache, _tailscale_feeds_cache_at, _tailscale_feeds_cache_lock, FEED_LAST_POST_TTL_SECONDS, updating_feeds, updating_feeds_lock, _PerUserDict, # noqa: F401 -- re-exported: routes/integrations_youtube.py and tests still do `main._PerUserDict` _ad_asset_hashes_cache, _ad_asset_hashes_lock, _autofetch_jobs, _autofetch_jobs_lock, _AUTOFETCH_JOB_STALE_S, _has_manual_tags_cache, _has_manual_tags_lock, HAS_MANUAL_TAGS_CACHE_TTL_SECONDS, _meta_structure_cache, _meta_structure_lock, _problematic_feeds_cache, _problematic_feeds_cache_lock, PROBLEMATIC_FEEDS_CACHE_TTL_SECONDS, PROBLEMATIC_FEEDS_LIMIT, _refetch_jobs, _refetch_jobs_lock, feed_title_map_cache, feed_title_map_cache_lock, FEED_TITLE_MAP_CACHE_TTL_SECONDS, tag_counts_cache, tag_counts_cache_lock, TAG_COUNTS_CACHE_TTL_SECONDS, unread_counts_cache, unread_counts_cache_lock, UNREAD_COUNTS_CACHE_TTL_SECONDS, _bump_unread_counts_generation, get_unread_counts_generation, invalidate_unread_counts_cache, try_start_unread_refresh, clear_unread_refresh_inflight, next_refresh_rotation_offset, check_manual_refresh_cooldown, unread_counts_compute_lock, ) BASE_DIR = Path(__file__).resolve().parent # Dedicated app logger. Previously this piggybacked on "uvicorn.error", but # uvicorn owns that logger and (depending on startup/rollover timing) can leave # it not propagating to our root file handler — silently dropping every app log # line ([maintenance], [refresh], …) even though the code ran. "lectio" is our # own namespace that nothing else reconfigures, so it always reaches root. LOGGER = logging.getLogger("lectio") LOGGER.setLevel(logging.INFO) # Refresh and page-fetch threads keep this one process's GIL busy, and a request makes many short DB calls that each # have to win it back (5ms per turn by default) -- measured 2026-10-02 as the same steps taking ~50ms at rest and 1-4s # during a refresh burst. A shorter interval hands it back sooner; throughput cost is small. try: sys.setswitchinterval(float(os.getenv("LECTIO_GIL_SWITCH_INTERVAL", "0.001"))) except ValueError: pass class _ReaderNonFatalParseWarningFilter(logging.Filter): """Filter known non-fatal feed parsing warnings from `reader` logs.""" def filter(self, record: logging.LogRecord) -> bool: message = record.getMessage() if not message.startswith("parse "): return True suppressed_markers = ( "got NonXMLContentType(", "got CharacterEncodingOverride(", ) return not any(marker in message for marker in suppressed_markers) class _ReaderUpdateFetchErrorFilter(logging.Filter): """Demote reader's per-feed fetch/parse failures from ERROR+traceback to a one-line WARNING. reader logs every failed ``update_feed`` at ERROR with a full stack trace via the ``reader.update`` logger — but Lectio already catches each one, applies backoff, records ``feed_failure_state.last_error``, and surfaces it in the problematic-feeds panel. So a routine 429/403/timeout/dead-feed produced a scary red traceback in the log (and the new Admin → Logs tab) for a condition that's fully handled. Keep the one-line summary as a WARNING but drop the redundant traceback; genuinely unexpected reader errors stay loud (ERROR).""" # Signatures of expected, Lectio-handled fetch/parse failures. _EXPECTED = ( "bad http status code", # any HTTP error status (429/403/404/410/5xx) "too many requests", "forbidden", "not found", "gone", "service unavailable", "timed out", "timeout", "connecttimeout", "readtimeout", "connectionerror", "connection reset", "connection aborted", "name or service not known", "temporary failure in name resolution", "certificate", "ssl", "tlsv1", "nonxmlcontenttype", "no parser", "document declared as", "not well-formed", ) def filter(self, record: logging.LogRecord) -> bool: if record.levelno < logging.ERROR: return True message = record.getMessage().lower() if "status=error" not in message and "updating feed" not in message: return True if any(marker in message for marker in self._EXPECTED): record.levelno = logging.WARNING record.levelname = "WARNING" record.exc_info = None record.exc_text = None return True def _configure_reader_logging() -> None: if os.getenv("LECTIO_SUPPRESS_READER_PARSE_WARNINGS", "1") != "1": return logging.getLogger("reader").addFilter(_ReaderNonFatalParseWarningFilter()) logging.getLogger("reader.update").addFilter(_ReaderUpdateFetchErrorFilter()) # Query-string secrets that reach the log in request URLs: httpx logs every outgoing URL (the YouTube Data API takes its key as # ``&key=``, signed image URLs carry a signature) and the access log records incoming ones (OAuth callbacks carry ``code``/``state``). _LOG_SECRET_RE = re.compile( r"([?&](?:key|api_key|access_token|refresh_token|client_secret|token|code|state|password|jwt|sig|signature" r"|x-amz-signature|x-amz-security-token)=)[^&\s\"']+", re.IGNORECASE, ) class _RedactingFormatter(logging.Formatter): """Log formatter that masks URL query-string secrets before the line is written anywhere.""" def format(self, record: logging.LogRecord) -> str: return _LOG_SECRET_RE.sub(r"\1", super().format(record)) def _configure_persistent_logging() -> None: """Attach a rotating file handler when LECTIO_LOG_DIR is set. Defaults: 5 MB per file, 5 backups (LECTIO_LOG_MAX_BYTES, LECTIO_LOG_BACKUPS). This is the durable record; :func:`_configure_console_logging` mirrors the same stream to the console. Without LECTIO_LOG_DIR only the console handler runs. """ log_dir_str = os.getenv("LECTIO_LOG_DIR", "").strip() if not log_dir_str: return from logging.handlers import RotatingFileHandler log_dir = Path(log_dir_str) log_dir.mkdir(parents=True, exist_ok=True) max_bytes = int(os.getenv("LECTIO_LOG_MAX_BYTES", str(5 * 1024 * 1024))) backup_count = int(os.getenv("LECTIO_LOG_BACKUPS", "5")) handler = RotatingFileHandler( log_dir / "lectio.log", maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8", ) handler.setFormatter(_RedactingFormatter("%(asctime)s %(levelname)s %(name)s: %(message)s")) # Attach to the root logger so it captures everything (uvicorn, reader, app). root = logging.getLogger() if root.level > logging.INFO or root.level == logging.NOTSET: root.setLevel(logging.INFO) root.addHandler(handler) def _configure_console_logging() -> None: """Mirror everything the app logs to the console, so `docker compose logs` shows it. uvicorn installs handlers on its own loggers with propagate=False, so its startup lines appear on the console while every app, reader and library log record propagates to the root logger — which, in the container, owns only the rotating file handler under LECTIO_LOG_DIR. The result was that all app logging was invisible to `docker compose logs` and readable only from inside the container. That cost hours during the 2026-08-18 restart hunt: the scheduler watchdog logged two ERROR lines before every exit and the container log showed nothing but a silent restart. Writes to stderr, which is where uvicorn's own handlers write, so the interleaving stays sane. The check below excludes FileHandler because RotatingFileHandler is itself a StreamHandler subclass. """ root = logging.getLogger() for existing in root.handlers: if isinstance(existing, logging.StreamHandler) and not isinstance(existing, logging.FileHandler): if not isinstance(existing.formatter, _RedactingFormatter): existing.setFormatter(_RedactingFormatter(existing.formatter._fmt if existing.formatter else None)) return handler = logging.StreamHandler() handler.setFormatter(_RedactingFormatter("%(asctime)s %(levelname)s %(name)s: %(message)s")) # INFO by default, because the whole point is that app logs become visible. But loosening the root # level is a decision about someone else's process, so LECTIO_LOG_LEVEL takes precedence when set — # a deployment that wants a quieter console says so instead of being overruled. wanted = os.getenv("LECTIO_LOG_LEVEL", "").strip().upper() if wanted: root.setLevel(getattr(logging, wanted, logging.INFO)) elif root.level > logging.INFO or root.level == logging.NOTSET: root.setLevel(logging.INFO) root.addHandler(handler) def _configure_access_log_filter() -> None: # Kept for backward-compat with previous wiring; the actual access log is # now emitted by _AccessLogMiddleware (uvicorn's --no-access-log disables # the built-in one whose filters we couldn't reliably hook). pass def _attach_pending_access_filter() -> None: pass _configure_reader_logging() _configure_console_logging() _configure_persistent_logging() _configure_access_log_filter() def _load_local_env(env_path: str | Path | None = None) -> None: """Load simple KEY=VALUE pairs from a .env file into os.environ if not already set. This is intentionally minimal (no interpolation) and avoids adding a dependency. """ candidates = [] if env_path: candidates.append(Path(env_path)) candidates.append(BASE_DIR / ".env") candidates.append(Path.cwd() / ".env") for p in candidates: try: if not p or not p.exists(): continue with p.open("r", encoding="utf-8") as fh: for raw in fh: line = raw.strip() if not line or line.startswith("#"): continue if "=" not in line: continue key, val = line.split("=", 1) key = key.strip() val = val.strip() if (val.startswith('"') and val.endswith('"')) or (val.startswith("'") and val.endswith("'")): val = val[1:-1] # Do not overwrite existing environment vars if key and os.getenv(key) is None: os.environ[key] = val break except Exception: continue # Load local .env early so os.getenv() below can pick up values provided by the developer _load_local_env() # Default to BASE_DIR/data (matching the Docker ./data:/data mapping) so a local # run without LECTIO_DATA_DIR set writes DBs and service dirs (scraped-feeds/, # deviantart-feeds/) into ./data instead of polluting the repo root. # expanduser so a tilde path (LECTIO_DATA_DIR=~/.lectio) resolves against $HOME # rather than being treated as a literal "./~" segment. _data_dir_env = os.getenv("LECTIO_DATA_DIR") DATA_DIR = Path(os.path.expanduser(_data_dir_env) if _data_dir_env else BASE_DIR / "data").resolve() DATA_DIR.mkdir(parents=True, exist_ok=True) THUMB_DB_PATH = DATA_DIR / "lectio_thumb_cache.sqlite" # Global, content-addressed (source-URL hash -> original image bytes) cache for the # /api/img proxy. Like the thumb cache, it holds no per-user data and is NOT routed # through tenancy. See ensure_img_cache_schema / api_img_proxy. IMG_CACHE_DB_PATH = DATA_DIR / "lectio_img_cache.sqlite" # Global, content-addressed (video_id -> duration): a fact about the video, not # per-user, so it's shared — a user without a YouTube key still gets durations # another user's key already fetched. Like the thumb cache, NOT routed through tenancy. YT_DURATION_DB_PATH = DATA_DIR / "lectio_yt_durations.sqlite" # Global account registry (NOT per-user, NOT routed through tenancy): one users # table for the whole instance. AUTH_DB_PATH = DATA_DIR / "lectio_auth.sqlite" THUMB_CACHE_DIR = DATA_DIR / "thumb_cache" # legacy on-disk cache; entries migrate lazily on access # Bind the tenancy resolver: per-user DBs live under DATA_DIR/users/{uid}/. The thumb # cache (THUMB_DB_PATH) is intentionally NOT routed through tenancy — it is a # content-addressed global cache shared across all users. See services/tenancy.py. tenancy.configure(data_dir=DATA_DIR) ROOT_FOLDER_NAME = "All Feeds" _LECTIO_FOLDER_NAME = "_Lectio" # Virtual, derived folder pinned to the bottom of the sidebar that holds every # subscribed feed with no folder_feeds row (e.g. feeds imported via OPML/reader # migration that never got a folder). It has no DB row; its id is a negative # sentinel so it never collides with real (positive) folder ids, and its # membership is computed per-render as (all reader feeds − foldered feeds). UNCATEGORIZED_FOLDER_ID = -1 UNCATEGORIZED_FOLDER_NAME = "Uncategorized" scraper_service.init(DATA_DIR) deviantart_service.init(DATA_DIR) devto_service.init(DATA_DIR) def _da_seed_lead_image(feed_url: str, entry_id: str, image_url: str) -> None: """Sink for the DeviantArt service: store the API image as the entry's lead image (DB + live cache) unless one is already cached (preserve customizations).""" try: if not lead_image_service.get_cached_entry_thumbnail(feed_url, entry_id, ""): lead_image_service.store_entry_lead_image(feed_url, entry_id, image_url) except Exception: LOGGER.exception("[deviantart] failed to seed lead image for %s", entry_id) deviantart_service.set_lead_image_sink(_da_seed_lead_image) def _devto_seed_lead_image(feed_url: str, entry_id: str, image_url: str) -> None: """Sink for the dev.to service: store the API cover image as the entry's lead image (DB + live cache) unless one is already cached (preserve customizations).""" try: if not lead_image_service.get_cached_entry_thumbnail(feed_url, entry_id, ""): lead_image_service.store_entry_lead_image(feed_url, entry_id, image_url) except Exception: LOGGER.exception("[devto] failed to seed lead image for %s", entry_id) devto_service.set_lead_image_sink(_devto_seed_lead_image) DEFAULT_AUTO_REFRESH_MINUTES = 60 MIN_AUTO_REFRESH_MINUTES = 5 MANUAL_REFRESH_COOLDOWN_SECONDS = 60 FAILED_FEED_BACKOFF_BASE_SECONDS = 60 FAILED_FEED_BACKOFF_MAX_SECONDS = 60 * 60 * 24 AUTO_REFRESH_SETTING_KEY = "auto_refresh_minutes" SORT_BY_SETTING_KEY = "sort_by" SORT_DIR_SETTING_KEY = "sort_dir" # Feeds and Saved remember their sort SEPARATELY. One shared pair meant picking # an order in Saved silently changed Feeds, and vice versa — the two are # different jobs (a publish-date backlog vs a to-do pile), so they want # different orders. The unprefixed keys stay Feeds' so existing installs keep # their remembered value. SAVED_SORT_BY_SETTING_KEY = "saved_sort_by" SAVED_SORT_DIR_SETTING_KEY = "saved_sort_dir" def sort_setting_keys(star_only: bool) -> tuple[str, str]: """The (sort_by, sort_dir) setting keys for the active scope.""" if star_only: return SAVED_SORT_BY_SETTING_KEY, SAVED_SORT_DIR_SETTING_KEY return SORT_BY_SETTING_KEY, SORT_DIR_SETTING_KEY GLOBAL_NOTE_SETTING_KEY = "global_note" PROBLEMATIC_FEEDS_LAST_VIEWED_AT_SETTING_KEY = "problematic_feeds_last_viewed_at" YOUTUBE_SYNC_LAST_AT_KEY = "youtube_sync_last_at" YOUTUBE_SYNC_LAST_RESULT_KEY = "youtube_sync_last_result" EMAIL_TO_SETTING_KEY = "email_to" EMAIL_BCC_SETTING_KEY = "email_bcc_address" PROFILE_NAME_SETTING_KEY = "profile_name" PROFILE_EMAIL_SETTING_KEY = "profile_email" SETTING_TZ_DISPLAY = "tz_display" SETTING_PORTRAIT_IMG_MAX_WIDTH = "portrait_img_max_width" # Per-user: Ctrl/Cmd-click or middle-click on a post title (opens the article in a new tab without opening it in the reader) marks the # post read. Off ("0") by default: those clicks leave it unread. SETTING_MARK_READ_ON_NEW_TAB = "mark_read_on_new_tab" # Per-user: rewrite every third-party in the article pane body to # /api/img?u=… instead of loading it directly. Off ("0") by default — raises # /api/img traffic and cache size, and drops srcset (so high-DPI screens fetch # the proxied single-resolution src). Read Mode always does this regardless of # this setting (see proxy_all_body_images). SETTING_PROXY_BODY_IMAGES = "proxy_body_images" # Outbound-fetch SOCKS5/HTTP proxy (e.g. a gluetun VPN container). URL is # admin-only (a non-admin choosing an arbitrary proxy target is SSRF-adjacent); # mode is per-user-overridable — "" means inherit the instance default. See # get_proxy_url / get_proxy_mode. SETTING_PROXY_URL = "proxy_url" SETTING_PROXY_MODE = "proxy_mode" # "off" / "as_needed" / "always" # Last-resort outbound proxy (e.g. a Tailscale exit node reaching the real home # IP) — a different trust tier from SETTING_PROXY_URL: real exposure cost, not a # disposable datacenter NAT. Rides the SAME proxy_mode as the primary proxy # rather than its own mode (Josh's own framing: mirror the existing pattern, # not invent a new one) — its actual gate is simply being unconfigured by # default, same as the primary proxy. Only ever reached via as_needed's # per-feed escalation, one rung past the primary proxy, never via "always" # (that would route every fetch through the home IP, which "last resort" rules # out). URL is admin-only for the same SSRF-adjacent reason as SETTING_PROXY_URL. SETTING_TAILSCALE_URL = "tailscale_url" # FlareSolverr (real headless Chrome, purpose-built to solve Cloudflare/DDoS- # Guard style challenges) — one rung past the primary proxy, before the # last-resort backend above. Rides the same proxy_mode too. No token/auth (an # internal-network-only container); zero personal exposure — always stacked # with the primary proxy (SETTING_PROXY_URL) when one is configured, never # with the last-resort one, since it doesn't need the home IP to work and # spending it here would defeat "last resort". Gated narrower than the other # two tiers: only a real challenge page (bot_challenge.FeedBlockedError, see # FeedRefreshService._is_bot_challenge), not any refusal — a plain IP-block # 403 wants a different exit IP, not a browser solving a challenge that was # never served. SETTING_FLARESOLVERR_URL = "flaresolverr_url" SETTING_MAINTENANCE_HOUR = "maintenance_hour" SETTING_IMG_CACHE_DAYS = "img_cache_days" SETTING_IMG_CACHE_MAX_DIM = "img_cache_max_dim" SETTING_IMG_TARGET_BYTES = "img_target_bytes" SETTING_YT_API_KEY = "yt_api_key" SETTING_YT_CHANNEL_ID = "yt_channel_id" SETTING_YT_FOLDER_NAME = "yt_folder_name" # Per-user: use the standard youtube.com embed host (account features — Share / # Watch Later — work) instead of the privacy-enhanced youtube-nocookie.com. Off # (privacy) by default; the standard host sets YouTube cookies. "1"/"0". SETTING_YT_EMBED_ACCOUNT_FEATURES = "yt_embed_account_features" # Global per-user toggle: auto-mark YouTube Shorts read across ALL YouTube feeds at # refresh, regardless of the per-feed hide_shorts pref. Off ("0") by default. SETTING_YT_HIDE_SHORTS_GLOBAL = "yt_hide_shorts_global" # Global per-user toggle: don't show YouTube videos that haven't premiered yet # (liveBroadcastContent == "upcoming") across ALL YouTube feeds, regardless of # the per-feed hide_unpremiered pref. Off ("0") by default. SETTING_YT_HIDE_UNPREMIERED_GLOBAL = "yt_hide_unpremiered_global" # Global per-user toggle: auto-mark members-only videos read across ALL YouTube # feeds at refresh, regardless of the per-feed hide_members_only pref. Off # ("0") by default. Detection has a real per-video cost (a watch-page fetch — # unlike hide_shorts/hide_unpremiered, membership status isn't in the Data API # response already fetched for duration/premiere status), so it only runs for # entries that have never been checked before (see youtube_video_duration's # members_only column). SETTING_YT_HIDE_MEMBERS_ONLY_GLOBAL = "yt_hide_members_only_global" # Global per-user toggle: hide a webcomic strip a publisher has locked behind a # supporter tier (cad-comic.com and the like) until its own stated unlock date, # across ALL webcomic feeds, regardless of the per-feed hide_locked_comics pref. # Not YouTube-specific — see LeadImageService.check_and_cache_webcomic_lock. # Off ("0") by default. SETTING_HIDE_LOCKED_COMICS_GLOBAL = "hide_locked_comics_global" # Daily YouTube Data API quota cap (units). Google's default is 10,000/day; make it # a setting in case a higher quota is granted. SETTING_YT_QUOTA_CAP = "yt_quota_cap" YT_QUOTA_DEFAULT_CAP = 10000 # Watermark for the auto-add-to-playlist rule's "what's new since I last # checked" scan — see _run_youtube_playlist_rules_after_refresh. SETTING_YT_PLAYLIST_AUTO_LAST_CHECK = "yt_playlist_auto_last_check" # "On star, also send to…" — per-user destinations fired when an article is starred. SETTING_STAR_SEND_INSTAPAPER = "star_send_instapaper" # "1"/"0" SETTING_STAR_SEND_YT_PLAYLIST = "star_send_yt_playlist" # playlist id ("" = off) SETTING_STAR_SEND_YT_PLAYLIST_TITLE = "star_send_yt_playlist_title" SETTING_STAR_SEND_EMAIL = "star_send_email" # address ("" = off) SETTING_RESEND_API_KEY = "resend_api_key" SETTING_EMAIL_FROM = "email_from" SETTING_INSTAPAPER_USERNAME = "instapaper_username" SETTING_INSTAPAPER_PASSWORD = "instapaper_password" SETTING_DEVIANTART_CLIENT_ID = "deviantart_client_id" SETTING_DEVIANTART_CLIENT_SECRET = "deviantart_client_secret" SETTING_DEVIANTART_ACCESS_TOKEN = "deviantart_access_token" SETTING_DEVIANTART_REFRESH_TOKEN = "deviantart_refresh_token" SETTING_DEVIANTART_TOKEN_EXPIRES_AT = "deviantart_token_expires_at" SETTING_DEVIANTART_USERNAME = "deviantart_username" SETTING_DEVIANTART_OAUTH_STATE = "deviantart_oauth_state" SETTING_DEVIANTART_OAUTH_VERIFIER = "deviantart_oauth_verifier" SETTING_DEVIANTART_FOLDER_NAME = "deviantart_folder_name" SETTING_DEVIANTART_SYNC_STATUS = "deviantart_sync_status" # Structured companion to the status string: JSON {failed:[{username,error}], unwatched:[username]} # so the Settings UI can list the failed/unwatched artists as profile links instead # of telling the user to "see logs" (which they have no access to). SETTING_DEVIANTART_SYNC_DETAIL = "deviantart_sync_detail" # Set "1" the moment a sync finds a NEWLY-unwatched (now auto-paused) artist; # cleared only when the DeviantArt Integrations sub-tab is actually opened — # drives the Integrations/DeviantArt tab dots, deliberately separate from # whether the unwatched list itself is empty (that list stays populated for # Josh to act on "at some point or not"; this flag is just "there's something # new in it you haven't looked at yet"). SETTING_DEVIANTART_UNWATCHED_DIRTY = "deviantart_unwatched_dirty" SETTING_YT_OAUTH_CLIENT_ID = "yt_oauth_client_id" SETTING_YT_OAUTH_CLIENT_SECRET = "yt_oauth_client_secret" SETTING_YT_OAUTH_ACCESS_TOKEN = "yt_oauth_access_token" SETTING_YT_OAUTH_REFRESH_TOKEN = "yt_oauth_refresh_token" SETTING_YT_OAUTH_TOKEN_EXPIRES_AT = "yt_oauth_token_expires_at" SETTING_YT_OAUTH_STATE = "yt_oauth_state" # Shared-instance YouTube OAuth creds stored in the admin's app_settings. SETTING_SHARED_YT_OAUTH_CLIENT_ID = "shared_yt_oauth_client_id" SETTING_SHARED_YT_OAUTH_CLIENT_SECRET = "shared_yt_oauth_client_secret" # Pinterest outbound (save / pin an article to a board). SETTING_PINTEREST_OAUTH_CLIENT_ID = "pinterest_oauth_client_id" SETTING_PINTEREST_OAUTH_CLIENT_SECRET = "pinterest_oauth_client_secret" SETTING_PINTEREST_OAUTH_ACCESS_TOKEN = "pinterest_oauth_access_token" SETTING_PINTEREST_OAUTH_REFRESH_TOKEN = "pinterest_oauth_refresh_token" SETTING_PINTEREST_OAUTH_TOKEN_EXPIRES_AT = "pinterest_oauth_token_expires_at" SETTING_PINTEREST_OAUTH_STATE = "pinterest_oauth_state" # Shared-instance Pinterest OAuth creds stored in the admin's app_settings. SETTING_SHARED_PINTEREST_OAUTH_CLIENT_ID = "shared_pinterest_oauth_client_id" SETTING_SHARED_PINTEREST_OAUTH_CLIENT_SECRET = "shared_pinterest_oauth_client_secret" # Inoreader migration — OAuth tokens and import checkpoint. SETTING_INOREADER_CLIENT_ID = "inoreader_client_id" SETTING_INOREADER_CLIENT_SECRET = "inoreader_client_secret" SETTING_INOREADER_ACCESS_TOKEN = "inoreader_access_token" SETTING_INOREADER_REFRESH_TOKEN = "inoreader_refresh_token" SETTING_INOREADER_TOKEN_EXPIRES_AT = "inoreader_token_expires_at" SETTING_INOREADER_OAUTH_STATE = "inoreader_oauth_state" SETTING_INOREADER_IMPORT_STATE = "inoreader_import_state" SETTING_INOREADER_EXPORT_DIR = "inoreader_export_dir" # server-side path to JSON export files # Miniflux migration (FROM a Miniflux instance — distinct from the Miniflux API compat server). SETTING_MINIFLUX_IMPORT_URL = "miniflux_import_url" SETTING_MINIFLUX_IMPORT_TOKEN = "miniflux_import_token" SETTING_MINIFLUX_IMPORT_STATE = "miniflux_import_state" # FreshRSS migration. SETTING_FRESHRSS_URL = "freshrss_url" SETTING_FRESHRSS_USERNAME = "freshrss_username" SETTING_FRESHRSS_PASSWORD = "freshrss_password" SETTING_FRESHRSS_IMPORT_STATE = "freshrss_import_state" # tt-rss migration. SETTING_TTRSS_URL = "ttrss_url" SETTING_TTRSS_USERNAME = "ttrss_username" SETTING_TTRSS_PASSWORD = "ttrss_password" SETTING_TTRSS_IMPORT_STATE = "ttrss_import_state" # Reddit OAuth integration. SETTING_REDDIT_CLIENT_ID = "reddit_client_id" SETTING_REDDIT_CLIENT_SECRET = "reddit_client_secret" SETTING_REDDIT_ACCESS_TOKEN = "reddit_oauth_access_token" SETTING_REDDIT_REFRESH_TOKEN = "reddit_oauth_refresh_token" SETTING_REDDIT_TOKEN_EXPIRES_AT = "reddit_oauth_token_expires_at" SETTING_REDDIT_OAUTH_STATE = "reddit_oauth_state" SETTING_REDDIT_USERNAME = "reddit_username" SETTING_SHARED_REDDIT_CLIENT_ID = "shared_reddit_client_id" SETTING_SHARED_REDDIT_CLIENT_SECRET = "shared_reddit_client_secret" SETTING_STAR_SEND_REDDIT_SUBREDDIT = "star_send_reddit_subreddit" # Instance tuning settings (admin-only, stored in admin's app_settings). SETTING_FETCH_HISTORY_MAX_AGE_DAYS = "fetch_history_max_age_days" # Auto-read age (days; 0 = off) at instance and account level. Each level below can only shorten it: folder, then feed. SETTING_AUTO_READ_DAYS_INSTANCE = "auto_read_days_instance" SETTING_AUTO_READ_DAYS_USER = "auto_read_days_user" SETTING_TOMBSTONE_SWEEP_DAYS = "tombstone_sweep_days" SETTING_LOGIN_MAX_FAILURES = "login_max_failures" SETTING_LOGIN_WINDOW_SECONDS = "login_window_seconds" SETTING_DEFAULT_AUTO_REFRESH_MINUTES = "default_auto_refresh_minutes" # Quire outbound (turn an article into a task in a chosen Quire project). SETTING_QUIRE_CLIENT_ID = "quire_client_id" SETTING_QUIRE_CLIENT_SECRET = "quire_client_secret" SETTING_QUIRE_ACCESS_TOKEN = "quire_access_token" SETTING_QUIRE_REFRESH_TOKEN = "quire_refresh_token" SETTING_QUIRE_TOKEN_EXPIRES_AT = "quire_token_expires_at" SETTING_QUIRE_OAUTH_STATE = "quire_oauth_state" SETTING_QUIRE_USERNAME = "quire_username" SETTING_QUIRE_PROJECT_OID = "quire_project_oid" # default destination project ("" = none) SETTING_QUIRE_PROJECT_NAME = "quire_project_name" # cached display name of that project SETTING_STAR_SEND_QUIRE = "star_send_quire" # "1"/"0" # Quire rate limits are per-organization, per-minute and per-hour (Free: 50/min, 200/hr). # These are Lectio's own sliding-window caps used to drive the usage meter + back off. SETTING_QUIRE_RATE_CAP_MIN = "quire_rate_cap_min" SETTING_QUIRE_RATE_CAP_HOUR = "quire_rate_cap_hour" SETTING_QUIRE_PLAN = "quire_plan" # detected plan name of the destination project's org (display only) QUIRE_RATE_DEFAULT_CAP_MIN = 50 QUIRE_RATE_DEFAULT_CAP_HOUR = 200 AUTO_REFRESH_OPTION_MINUTES = (0, 5, 15, 30, 60, 360, 720) SCHEDULER_POLL_SECONDS = 30 DEFAULT_SORT_BY = "post" DEFAULT_SORT_DIR = "asc" CHUNK_SIZE = 20 # feed_fetch_history retention configured via the Administration panel. READABILITY_USER_AGENT = "Lectio/0.1 (+https://localhost)" # Honest identifier for outbound fetches — names the app and links to the repo, # the good-citizen behavior some hosts (e.g. rachelbythebay.com) explicitly # reward. This is the DEFAULT for podcast discovery/borrow fetches. LECTIO_HONEST_USER_AGENT = "Lectio/0.1 (+https://github.com/joshg253/Lectio)" # Some podcast-host feeds (Buzzsprout, Libsyn, …) sit behind Cloudflare, which # 403s any non-browser UA. We escalate to this browser UA ONLY after the honest # request is actually refused (see _polite_safe_get) — never preemptively, so we # don't spoof hosts that are happy to serve Lectio. PODCAST_FETCH_USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36" # In-memory cache of domains known to have Cross-Origin-Resource-Policy restrictions. # Values: True = same-site/same-origin (proxy needed), False = no restriction. _CORP_DOMAIN_CACHE: dict[str, bool] = {} # Hosts that serve a "this image was hotlinked" placeholder when the request # carries a foreign Referer (or no same-origin signal). referrerpolicy="no-referrer" # defeats this for *fresh* image loads, but a browser that cached the placeholder # under the (unchanged) image URL keeps serving it. Routing these hosts' images # through the /api/img proxy gives the browser a new same-origin URL it hasn't # cached, and the server fetch carries no Referer — so the real image loads and # stays correct. Add a registrable domain (matches it and any subdomain). # wixmp.com serves DeviantArt images behind short-lived signed (?token=) URLs; # proxying caches the bytes server-side so the article image survives token expiry. # private-user-images.githubusercontent.com serves GitHub release/issue screenshots # behind short-lived (~5 min) JWT-signed URLs — same problem, same fix. # fabiensanglard.net is the *inverse* hotlink case: it 403s an image fetched with # no Referer (e.g. its .webp files) but serves it with a same-origin Referer. The # browser can't send fabiensanglard's own origin as Referer, so a direct # load fails (its .webp break in reader/web view while its .jpg loads); routing # through /api/img lets the proxy's same-origin-Referer retry (api_img_proxy) # fetch the real bytes. See ARCHITECTURE "Same-origin Referer". _HOTLINK_IMG_HOSTS: frozenset[str] = frozenset( {"nanolx.org", "wixmp.com", "private-user-images.githubusercontent.com", "fabiensanglard.net"} ) def _is_hotlink_img_host(netloc: str) -> bool: host = (netloc or "").split("@")[-1].split(":")[0].lower() return any(host == h or host.endswith("." + h) for h in _HOTLINK_IMG_HOSTS) MANUAL_TAG_KEY_PREFIX = "lectio.manual_tag." MAX_MANUAL_TAGS = 12 # Hard cap on how many of an entry's feed tags are carried to the page. RPS # ships 28 on a single post, so this is generous on purpose — the cost of one # extra chip is that it is ignored, and the cost of a missing one is that the # tag cannot be filtered on at all. MAX_FEED_TAG_SUGGESTIONS = 40 # How many chips the row shows before the "+N more" expander. Eight fits the # header without wrapping; the rest are one click away rather than absent. FEED_TAG_CHIPS_COLLAPSED = 8 TAG_VALUE_PATTERN = re.compile(r"^[A-Za-z0-9_.#+][A-Za-z0-9_.#+-]{0,31}$") def _static_asset_version() -> str: """Cache-buster for every ?v=-versioned static asset. Hashes ALL css/js under static/ rather than a hand-kept list. The list went stale exactly as you would expect: offline-probe.js and js/cleanup.js are both served with ?v={{ static_asset_version }} and neither was in it, so editing them left the version unchanged and browsers — and the service worker, which caches /static by design — kept serving the previous file. Deploying a fix and watching the old copy run is a bad way to spend an hour. `.webmanifest` is in the set for the same reason: it is served with the same ?v= and the same immutable, year-long Cache-Control, so a manifest edit that did not move the digest would be pinned in the browser indefinitely. Sorted so the digest is stable across filesystems. """ try: _static = Path(__file__).parent / "static" paths = sorted(p for p in _static.rglob("*") if p.is_file() and p.suffix in (".css", ".js", ".webmanifest")) combined = b"".join(p.read_bytes() for p in paths) return hashlib.md5(combined).hexdigest()[:10] except Exception: return "dev" STATIC_ASSET_VERSION = os.getenv("LECTIO_ASSET_VERSION") or _static_asset_version() REFRESH_DEBUG_ENABLED = os.getenv("LECTIO_REFRESH_DEBUG", "0") == "1" # Gates the slow-SQL timing/progress-step instrumentation (Plan.md Tier 1 # refresh-contention item) on both the meta and reader DB connections. Off by # default: a sqlite3 progress handler fires every 1000 VM instructions on # every statement, and the timing wrapper adds a perf_counter() pair per call # -- real, if small, overhead on every single query, all the time, not worth # paying outside an active investigation. Flip on when chasing a live stall. LECTIO_PERF_DEBUG = os.getenv("LECTIO_PERF_DEBUG", "0") == "1" def _env_int(name: str, default: int) -> int: try: return int(os.getenv(name, str(default))) except ValueError: return default # /api/img proxy cache tuning (instance-level; admins can override in the # Administration page, which takes precedence over these env fallbacks). # LECTIO_IMG_CACHE_DAYS last-accessed TTL in days; 0 = keep forever (default 90) # LECTIO_IMG_CACHE_MAX_DIM longest-side px to downscale stored images to (default 3840) _ENV_IMG_CACHE_DAYS = _env_int("LECTIO_IMG_CACHE_DAYS", 90) _ENV_IMG_CACHE_MAX_DIM = _env_int("LECTIO_IMG_CACHE_MAX_DIM", 3840) # Byte budget for a cached image. The dimension cap above is not a size cap — a # 4K RGBA PNG sits exactly at MAX_DIM and still weighs ~12 MB, which is then # served whole on every view. Over this, the image is re-encoded to WebP # (lossless first, so screenshots and line art keep their edges). 0 disables. _ENV_IMG_TARGET_BYTES = _env_int("LECTIO_IMG_TARGET_BYTES", 1_500_000) # Max width (px) for portrait (taller-than-wide) article images; 0 = disabled. _ENV_PORTRAIT_IMG_MAX_WIDTH = _env_int("LECTIO_PORTRAIT_IMG_MAX_WIDTH", 650) # Outbound-fetch proxy endpoint (e.g. socks5h://gluetun:1080). "" = none configured. _ENV_PROXY_URL = os.getenv("LECTIO_PROXY_URL", "").strip() # Last-resort outbound proxy (e.g. socks5h://tailscale:1080). "" = none configured. _ENV_TAILSCALE_URL = os.getenv("LECTIO_TAILSCALE_URL", "").strip() # FlareSolverr endpoint (e.g. http://flaresolverr:8191/v1). "" = none configured. _ENV_FLARESOLVERR_URL = os.getenv("LECTIO_FLARESOLVERR_URL", "").strip() # --- Scheduler resilience (see ARCHITECTURE "Refresh scheduler") --- # Feed fetches are strictly sequential, so one host that accepts a connection and # then goes quiet delays every feed behind it. These bound that. # LECTIO_FEED_CONNECT_TIMEOUT seconds to establish a feed connection # LECTIO_FEED_READ_TIMEOUT seconds to wait for any single read from a feed host FEED_CONNECT_TIMEOUT_SECONDS = float(_env_int("LECTIO_FEED_CONNECT_TIMEOUT", 10)) FEED_READ_TIMEOUT_SECONDS = float(_env_int("LECTIO_FEED_READ_TIMEOUT", 30)) # A read deadline is per-socket-read, not total: a host trickling one byte at a # time never trips it. The watchdog is the backstop — it watches for PROGRESS, # not elapsed time, because a full-library pass legitimately runs for an hour. # LECTIO_SCHEDULER_STALL_SECONDS no progress for this long => log loudly # LECTIO_SCHEDULER_STALL_RESTART_SECONDS no progress for this long => exit so the # container restarts; 0 disables SCHEDULER_STALL_SECONDS = _env_int("LECTIO_SCHEDULER_STALL_SECONDS", 600) SCHEDULER_STALL_RESTART_SECONDS = _env_int("LECTIO_SCHEDULER_STALL_RESTART_SECONDS", 1800) SCHEDULER_WATCHDOG_POLL_SECONDS = 60 DEBUG_MODE = os.getenv("LECTIO_DEBUG", "0") == "1" # Public base URL of this instance (e.g. https://lectio.example.com). # Required for WebSub: hubs need a reachable callback URL. Leave blank to disable WebSub. LECTIO_PUBLIC_URL = os.getenv("LECTIO_PUBLIC_URL", "").strip().rstrip("/") _FEVER_PASSWORD = os.getenv("LECTIO_FEVER_PASSWORD", "").strip() # --- Email (Resend) config — env vars are fallbacks; DB settings take precedence at runtime --- _ENV_RESEND_API_KEY = os.getenv("RESEND_API_KEY", "").strip() _ENV_RESEND_FROM = os.getenv("LECTIO_EMAIL_FROM", "").strip() # Seed value for the bootstrap admin's default recipient (then per-user). _ENV_RESEND_TO = os.getenv("LECTIO_EMAIL_TO", "").strip() def get_resend_api_key() -> str: # Instance-shared: the Resend account/key is owned at the instance level # (one verified domain), so the env value is the shared default for everyone. return get_runtime_setting(SETTING_RESEND_API_KEY, _ENV_RESEND_API_KEY) def get_resend_from() -> str: # Instance sending identity (admin-managed, like the Resend key): env value is # the shared default for everyone; an admin override lives in their settings. return get_runtime_setting(SETTING_EMAIL_FROM, _ENV_RESEND_FROM) def is_email_configured() -> bool: return bool(get_resend_api_key() and get_resend_from()) def is_instapaper_configured() -> bool: return bool(get_runtime_setting(SETTING_INSTAPAPER_USERNAME) and get_runtime_setting(SETTING_INSTAPAPER_PASSWORD)) # --- YouTube subscription sync config — env vars are fallbacks; DB settings take precedence --- _ENV_YT_API_KEY = os.getenv("YOUTUBE_API_KEY", "").strip() _ENV_YT_CHANNEL_ID = os.getenv("YOUTUBE_CHANNEL_ID", "").strip() _ENV_YT_FOLDER_NAME = os.getenv("YOUTUBE_FOLDER_NAME", "").strip() or "YouTube Subscriptions" # DeviantArt API creds — per-user DB settings take precedence; env is the fallback. _ENV_DEVIANTART_CLIENT_ID = os.getenv("DEVIANTART_CLIENT_ID", "").strip() _ENV_DEVIANTART_CLIENT_SECRET = os.getenv("DEVIANTART_CLIENT_SECRET", "").strip() # YouTube OAuth client creds are app-level (one registered Google app), not # per-user — only the resulting tokens are per-user. So these are read straight # from env for every user (sharing the app, never the tokens). _ENV_YT_OAUTH_CLIENT_ID = os.getenv("YOUTUBE_OAUTH_CLIENT_ID", "").strip() _ENV_YT_OAUTH_CLIENT_SECRET = os.getenv("YOUTUBE_OAUTH_CLIENT_SECRET", "").strip() # Pinterest OAuth client creds are app-level (one registered Pinterest app), same # as YouTube above — only the resulting tokens are per-user. _ENV_PINTEREST_OAUTH_CLIENT_ID = os.getenv("PINTEREST_OAUTH_CLIENT_ID", "").strip() _ENV_PINTEREST_OAUTH_CLIENT_SECRET = os.getenv("PINTEREST_OAUTH_CLIENT_SECRET", "").strip() # Quire API creds — per-user DB settings take precedence; env is the fallback # (same pattern as DeviantArt). Only the resulting tokens are ever per-user. _ENV_QUIRE_CLIENT_ID = os.getenv("QUIRE_CLIENT_ID", "").strip() _ENV_QUIRE_CLIENT_SECRET = os.getenv("QUIRE_CLIENT_SECRET", "").strip() # Inoreader migration OAuth creds — env is the fallback; DB settings take precedence. _ENV_INOREADER_CLIENT_ID = os.getenv("INOREADER_CLIENT_ID", "").strip() _ENV_INOREADER_CLIENT_SECRET = os.getenv("INOREADER_CLIENT_SECRET", "").strip() _yt_sync_hour_raw = os.getenv("YOUTUBE_SYNC_HOUR", "").strip() _ENV_MAINTENANCE_HOUR: int | None = None if _yt_sync_hour_raw: try: _h = int(_yt_sync_hour_raw) if 0 <= _h <= 23: _ENV_MAINTENANCE_HOUR = _h except ValueError: pass def get_yt_api_key() -> str: return get_runtime_setting(SETTING_YT_API_KEY, _ENV_YT_API_KEY) def get_yt_channel_id() -> str: return get_runtime_setting(SETTING_YT_CHANNEL_ID, _ENV_YT_CHANNEL_ID) def get_yt_folder_name() -> str: return get_runtime_setting(SETTING_YT_FOLDER_NAME, _ENV_YT_FOLDER_NAME) or "YouTube Subscriptions" def youtube_embed_account_features_enabled() -> bool: """Per-user: True if YouTube embeds should use the standard youtube.com host (Share / Watch Later) instead of the privacy-enhanced -nocookie host. Privacy-enhanced is the default.""" return get_runtime_setting(SETTING_YT_EMBED_ACCOUNT_FEATURES, "0") == "1" def youtube_hide_shorts_global() -> bool: """Per-user: auto-mark Shorts read on ALL YouTube feeds at refresh (overrides the per-feed pref). Off by default.""" return get_runtime_setting(SETTING_YT_HIDE_SHORTS_GLOBAL, "0") == "1" def youtube_hide_unpremiered_global() -> bool: """Per-user: hide not-yet-premiered videos on ALL YouTube feeds (overrides the per-feed pref). Off by default.""" return get_runtime_setting(SETTING_YT_HIDE_UNPREMIERED_GLOBAL, "0") == "1" def youtube_hide_members_only_global() -> bool: """Per-user: auto-mark members-only videos read on ALL YouTube feeds at refresh (overrides the per-feed pref). Off by default.""" return get_runtime_setting(SETTING_YT_HIDE_MEMBERS_ONLY_GLOBAL, "0") == "1" def hide_locked_comics_global() -> bool: """Per-user: hide a supporter-locked webcomic strip on ALL webcomic feeds until its own stated unlock date (overrides the per-feed pref). Off by default.""" return get_runtime_setting(SETTING_HIDE_LOCKED_COMICS_GLOBAL, "0") == "1" def _pacific_today() -> str: """Today's date (YYYY-MM-DD) in US/Pacific — the timezone YouTube resets quota on.""" import datetime as _dt try: from zoneinfo import ZoneInfo return _dt.datetime.now(ZoneInfo("America/Los_Angeles")).strftime("%Y-%m-%d") except Exception: # Fallback: fixed -08:00 offset (good enough for a day-bucket if tzdata is absent). return (_dt.datetime.now(_dt.timezone.utc) - _dt.timedelta(hours=8)).strftime("%Y-%m-%d") def youtube_quota_cap() -> int: try: return max(1, int(get_runtime_setting(SETTING_YT_QUOTA_CAP, str(YT_QUOTA_DEFAULT_CAP)) or YT_QUOTA_DEFAULT_CAP)) except TypeError, ValueError: return YT_QUOTA_DEFAULT_CAP def record_yt_quota_spend(units: int) -> None: """Add ``units`` to the current user's Pacific-day YouTube quota tally. Called by the YT services after each billed API call (sink wired in at startup).""" if units <= 0: return try: with get_meta_connection() as conn: conn.execute( "INSERT INTO yt_quota_spend (day, units) VALUES (?, ?) ON CONFLICT(day) DO UPDATE SET units = units + excluded.units", (_pacific_today(), int(units)), ) except Exception: LOGGER.debug("[yt-quota] failed to record %d units", units, exc_info=True) def mark_yt_quota_exhausted() -> None: """Snap today's tally to the cap after an actual quotaExceeded response.""" try: cap = youtube_quota_cap() with get_meta_connection() as conn: conn.execute( "INSERT INTO yt_quota_spend (day, units) VALUES (?, ?) ON CONFLICT(day) DO UPDATE SET units = MAX(units, excluded.units)", (_pacific_today(), cap), ) except Exception: LOGGER.debug("[yt-quota] failed to mark exhausted", exc_info=True) def get_yt_quota_spent_today() -> int: try: with get_meta_connection() as conn: row = conn.execute("SELECT units FROM yt_quota_spend WHERE day = ?", (_pacific_today(),)).fetchone() return int(row["units"]) if row else 0 except Exception: return 0 def get_yt_quota_status() -> dict: """Quota meter payload for the UI: spent / cap / remaining + a low/exhausted flag.""" cap = youtube_quota_cap() spent = get_yt_quota_spent_today() remaining = max(0, cap - spent) state = "exhausted" if remaining <= 0 else ("low" if remaining < 500 else "ok") return {"spent": spent, "cap": cap, "remaining": remaining, "state": state} def youtube_embed_host() -> str: """The YouTube embed host for the current user's privacy/features preference.""" return "www.youtube.com" if youtube_embed_account_features_enabled() else "www.youtube-nocookie.com" def invalidate_instance_setting_cache() -> None: with _instance_setting_cache_lock: _instance_setting_cache.clear() def _get_admin_instance_setting(key: str) -> str: """First enabled admin's stored value for *key* ("" if none), from any context. Reads via get_setting (DB-backed) rather than the in-memory cache alone, so it works in background threads before the admin's cache has ever loaded (e.g. the 3am maintenance check right after a container restart). """ now = time.monotonic() with _instance_setting_cache_lock: hit = _instance_setting_cache.get(key) if hit and now - hit[0] < _INSTANCE_SETTING_TTL_SECONDS: return hit[1] val = "" # Never crash the caller (login lockout checks run pre-auth; a # half-provisioned admin dir must degrade to the env fallback). try: for u in user_store.list_users(): if not u.get("is_admin") or u.get("disabled"): continue with tenancy.user_context(u["user_id"]): with get_meta_connection() as conn: v = get_setting(conn, key) if v: val = v break except Exception: LOGGER.debug("admin instance-setting lookup failed for %r", key, exc_info=True) with _instance_setting_cache_lock: _instance_setting_cache[key] = (now, val) return val def get_instance_setting(key: str, env_fallback: str = "") -> str: """Resolve an instance-level (admin-managed) setting from any context. Resolution: current context's own setting (when a user is bound) → first enabled admin's setting → env fallback. The admin tier is what makes Administration → Instance Config authoritative for background threads, pre-auth checks, and other users. """ val = get_cached_setting(key) if val: return val return _get_admin_instance_setting(key) or env_fallback def _get_shared_credential(key: str) -> str: """Read a shared-instance credential from the first admin user's settings. Shared-instance OAuth creds are stored in the admin's own app_settings under a namespaced key (e.g. shared_yt_oauth_client_id) and are used as a middle tier: per-user setting → shared-instance → env var. """ return _get_admin_instance_setting(key) def get_fetch_history_max_age_days() -> int: return int(get_instance_setting(SETTING_FETCH_HISTORY_MAX_AGE_DAYS) or 30) # Fixed per-feed row ceiling for feed_fetch_history — a backstop against a # pathologically-retried feed, not a user-facing setting. Normal cadence (even # hourly for the full max-age window) stays far under this. FETCH_HISTORY_ROW_CEILING = 5000 def get_tombstone_sweep_days() -> int: """Age (days) after which deleted-entry tombstones are swept by nightly maintenance; 0 disables sweeping. By then the entry has long left the publisher's feed window, so a refresh can no longer resurrect it.""" return int(get_instance_setting(SETTING_TOMBSTONE_SWEEP_DAYS) or 90) def get_login_max_failures() -> int: return int(get_instance_setting(SETTING_LOGIN_MAX_FAILURES) or 5) def get_login_window_seconds() -> int: return int(get_instance_setting(SETTING_LOGIN_WINDOW_SECONDS) or 300) def get_instance_default_auto_refresh() -> int: raw = int(get_instance_setting(SETTING_DEFAULT_AUTO_REFRESH_MINUTES) or DEFAULT_AUTO_REFRESH_MINUTES) return 0 if raw <= 0 else max(raw, MIN_AUTO_REFRESH_MINUTES) def get_youtube_oauth_credentials() -> tuple[str, str]: """YouTube OAuth client (client_id, client_secret). Resolution: per-user setting → shared-instance (admin's settings) → env var. """ cid = ( get_runtime_setting(SETTING_YT_OAUTH_CLIENT_ID) or _get_shared_credential(SETTING_SHARED_YT_OAUTH_CLIENT_ID) or _ENV_YT_OAUTH_CLIENT_ID ) secret = ( get_runtime_setting(SETTING_YT_OAUTH_CLIENT_SECRET) or _get_shared_credential(SETTING_SHARED_YT_OAUTH_CLIENT_SECRET) or _ENV_YT_OAUTH_CLIENT_SECRET ) return cid, secret def youtube_oauth_connected() -> bool: """True if the current user has connected their YouTube account.""" with get_meta_connection() as conn: return bool(get_setting(conn, SETTING_YT_OAUTH_REFRESH_TOKEN)) def get_youtube_oauth_token() -> str: """Return a valid user access token for playlist writes, refreshing if needed. Empty string if the user hasn't connected their YouTube account or the refresh failed (revoked / testing-mode 7-day expiry) — caller must prompt a reconnect. """ with get_meta_connection() as conn: access = get_setting(conn, SETTING_YT_OAUTH_ACCESS_TOKEN) or "" refresh = get_setting(conn, SETTING_YT_OAUTH_REFRESH_TOKEN) or "" try: expires_at = float(get_setting(conn, SETTING_YT_OAUTH_TOKEN_EXPIRES_AT) or 0) except TypeError, ValueError: expires_at = 0.0 if not refresh: return "" if access and time.time() < expires_at - 60: return access cid, secret = get_youtube_oauth_credentials() try: data = youtube_oauth_service.refresh_access_token(cid, secret, refresh) except Exception as exc: # noqa: BLE001 LOGGER.warning("[youtube-oauth] token refresh failed; reconnect required: %s", exc) return "" with get_meta_connection() as conn: set_setting(conn, SETTING_YT_OAUTH_ACCESS_TOKEN, data["access_token"]) if data.get("refresh_token"): set_setting(conn, SETTING_YT_OAUTH_REFRESH_TOKEN, data["refresh_token"]) set_setting(conn, SETTING_YT_OAUTH_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600)))) return data["access_token"] def get_pinterest_oauth_credentials() -> tuple[str, str]: """Pinterest OAuth client (client_id, client_secret). Resolution: per-user setting → shared-instance (admin's settings) → env var. """ cid = ( get_runtime_setting(SETTING_PINTEREST_OAUTH_CLIENT_ID) or _get_shared_credential(SETTING_SHARED_PINTEREST_OAUTH_CLIENT_ID) or _ENV_PINTEREST_OAUTH_CLIENT_ID ) secret = ( get_runtime_setting(SETTING_PINTEREST_OAUTH_CLIENT_SECRET) or _get_shared_credential(SETTING_SHARED_PINTEREST_OAUTH_CLIENT_SECRET) or _ENV_PINTEREST_OAUTH_CLIENT_SECRET ) return cid, secret def pinterest_oauth_connected() -> bool: """True if the current user has connected their Pinterest account.""" with get_meta_connection() as conn: return bool(get_setting(conn, SETTING_PINTEREST_OAUTH_REFRESH_TOKEN)) def get_pinterest_oauth_token() -> str: """Return a valid user access token for pin writes, refreshing if needed. Empty string if the user hasn't connected Pinterest or the refresh failed (revoked / expired) — caller must prompt a reconnect.""" with get_meta_connection() as conn: access = get_setting(conn, SETTING_PINTEREST_OAUTH_ACCESS_TOKEN) or "" refresh = get_setting(conn, SETTING_PINTEREST_OAUTH_REFRESH_TOKEN) or "" try: expires_at = float(get_setting(conn, SETTING_PINTEREST_OAUTH_TOKEN_EXPIRES_AT) or 0) except TypeError, ValueError: expires_at = 0.0 if not refresh: return "" if access and time.time() < expires_at - 60: return access cid, secret = get_pinterest_oauth_credentials() try: data = pinterest_oauth_service.refresh_access_token(cid, secret, refresh) except Exception as exc: # noqa: BLE001 LOGGER.warning("[pinterest-oauth] token refresh failed; reconnect required: %s", exc) return "" with get_meta_connection() as conn: set_setting(conn, SETTING_PINTEREST_OAUTH_ACCESS_TOKEN, data["access_token"]) if data.get("refresh_token"): set_setting(conn, SETTING_PINTEREST_OAUTH_REFRESH_TOKEN, data["refresh_token"]) set_setting(conn, SETTING_PINTEREST_OAUTH_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600)))) return data["access_token"] def get_reddit_credentials() -> tuple[str, str]: """Reddit OAuth client (client_id, client_secret). Resolution: per-user setting → shared-instance (admin's settings) → empty. No env-var fallback — credentials are DB-only for this integration. """ cid = get_runtime_setting(SETTING_REDDIT_CLIENT_ID) or _get_shared_credential(SETTING_SHARED_REDDIT_CLIENT_ID) or "" secret = get_runtime_setting(SETTING_REDDIT_CLIENT_SECRET) or _get_shared_credential(SETTING_SHARED_REDDIT_CLIENT_SECRET) or "" return cid, secret def reddit_connected() -> bool: """True if the current user has connected their Reddit account.""" with get_meta_connection() as conn: return bool(get_setting(conn, SETTING_REDDIT_REFRESH_TOKEN)) def get_reddit_user_token() -> str: """Return a valid Reddit access token, refreshing if needed. Empty string if not connected or refresh failed — caller must prompt reconnect. """ with get_meta_connection() as conn: access = get_setting(conn, SETTING_REDDIT_ACCESS_TOKEN) or "" refresh = get_setting(conn, SETTING_REDDIT_REFRESH_TOKEN) or "" try: expires_at = float(get_setting(conn, SETTING_REDDIT_TOKEN_EXPIRES_AT) or 0) except TypeError, ValueError: expires_at = 0.0 if not refresh: return "" if access and time.time() < expires_at - 60: return access cid, secret = get_reddit_credentials() if not cid or not secret: return "" try: data = reddit_service.refresh_access_token(cid, secret, refresh) except Exception as exc: # noqa: BLE001 LOGGER.warning("[reddit] token refresh failed; reconnect required: %s", exc) return "" with get_meta_connection() as conn: set_setting(conn, SETTING_REDDIT_ACCESS_TOKEN, data["access_token"]) if data.get("refresh_token"): set_setting(conn, SETTING_REDDIT_REFRESH_TOKEN, data["refresh_token"]) set_setting(conn, SETTING_REDDIT_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600)))) return data["access_token"] def _reddit_redirect_uri(request: Request) -> str: base = os.getenv("LECTIO_PUBLIC_URL", "").rstrip("/") if not base: base = str(request.base_url).rstrip("/") return f"{base}/integrations/reddit/oauth/callback" def get_inoreader_credentials() -> tuple[str, str]: """Inoreader OAuth credentials (client_id, client_secret). DB settings take precedence over env vars so any user can override them without restarting the container. """ cid = get_runtime_setting(SETTING_INOREADER_CLIENT_ID, _ENV_INOREADER_CLIENT_ID) secret = get_runtime_setting(SETTING_INOREADER_CLIENT_SECRET, _ENV_INOREADER_CLIENT_SECRET) return cid, secret def inoreader_connected() -> bool: with get_meta_connection() as conn: return bool(get_setting(conn, SETTING_INOREADER_REFRESH_TOKEN)) def get_inoreader_token() -> str: """Return a valid Inoreader access token, refreshing it if expired.""" with get_meta_connection() as conn: access = get_setting(conn, SETTING_INOREADER_ACCESS_TOKEN) or "" refresh = get_setting(conn, SETTING_INOREADER_REFRESH_TOKEN) or "" expires_at = float(get_setting(conn, SETTING_INOREADER_TOKEN_EXPIRES_AT) or 0) if not refresh: return "" if access and time.time() < expires_at - 60: return access cid, secret = get_inoreader_credentials() try: data = inoreader_service.refresh_access_token(cid, secret, refresh) except Exception as exc: # noqa: BLE001 LOGGER.warning("[inoreader] token refresh failed: %s", exc) if "invalid_grant" in str(exc): # Inoreader has permanently rejected this refresh token (not a transient # network/outage failure) — clear it so inoreader_connected() stops # reporting stale "Connected." and the UI offers Connect again. with get_meta_connection() as conn: set_setting(conn, SETTING_INOREADER_ACCESS_TOKEN, "") set_setting(conn, SETTING_INOREADER_REFRESH_TOKEN, "") set_setting(conn, SETTING_INOREADER_TOKEN_EXPIRES_AT, "") return "" with get_meta_connection() as conn: set_setting(conn, SETTING_INOREADER_ACCESS_TOKEN, data["access_token"]) if data.get("refresh_token"): set_setting(conn, SETTING_INOREADER_REFRESH_TOKEN, data["refresh_token"]) set_setting(conn, SETTING_INOREADER_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600)))) return data["access_token"] def get_deviantart_credentials() -> tuple[str, str]: """Per-user DeviantArt API credentials (client_id, client_secret). Stored per-user in app-settings; env vars are instance-wide fallbacks. """ cid = get_runtime_setting(SETTING_DEVIANTART_CLIENT_ID, _ENV_DEVIANTART_CLIENT_ID) secret = get_runtime_setting(SETTING_DEVIANTART_CLIENT_SECRET, _ENV_DEVIANTART_CLIENT_SECRET) return cid, secret def get_deviantart_user_token() -> str: """Return a valid user access token for watch-list ops, refreshing if expired. Empty string if the user hasn't connected their DeviantArt account. """ with get_meta_connection() as conn: access = get_setting(conn, SETTING_DEVIANTART_ACCESS_TOKEN) or "" refresh = get_setting(conn, SETTING_DEVIANTART_REFRESH_TOKEN) or "" try: expires_at = float(get_setting(conn, SETTING_DEVIANTART_TOKEN_EXPIRES_AT) or 0) except TypeError, ValueError: # A corrupt/manually-edited value must not break every token read; # treat it as expired so we fall through to a forced refresh. expires_at = 0.0 if not access: return "" if time.time() < expires_at - 60: return access if not refresh: return access # no refresh token; try the (possibly stale) access token cid, secret = get_deviantart_credentials() try: data = deviantart_service.refresh_access_token(cid, secret, refresh) except Exception as exc: # noqa: BLE001 # Refresh failed (expired/rotated refresh token, or the client_secret no # longer matches the app) — the session is dead; signal "reconnect needed" # rather than handing back a stale token that 401s deep in a sync. LOGGER.warning("[deviantart] token refresh failed; reconnect required: %s", exc) return "" with get_meta_connection() as conn: set_setting(conn, SETTING_DEVIANTART_ACCESS_TOKEN, data["access_token"]) if data.get("refresh_token"): set_setting(conn, SETTING_DEVIANTART_REFRESH_TOKEN, data["refresh_token"]) set_setting(conn, SETTING_DEVIANTART_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600)))) return data["access_token"] def get_quire_credentials() -> tuple[str, str]: """Per-user Quire API credentials (client_id, client_secret). Stored per-user in app-settings; env vars are instance-wide fallbacks. """ cid = get_runtime_setting(SETTING_QUIRE_CLIENT_ID, _ENV_QUIRE_CLIENT_ID) secret = get_runtime_setting(SETTING_QUIRE_CLIENT_SECRET, _ENV_QUIRE_CLIENT_SECRET) return cid, secret def is_quire_connected() -> bool: """True once the user has completed the Quire OAuth flow (has a token).""" return bool(get_runtime_setting(SETTING_QUIRE_ACCESS_TOKEN)) def quire_project_oid() -> str: """The chosen default destination project's OID ("" if none picked yet).""" return (get_runtime_setting(SETTING_QUIRE_PROJECT_OID) or "").strip() def quire_project_name() -> str: """Display name of the chosen default destination project ("" if none).""" return (get_runtime_setting(SETTING_QUIRE_PROJECT_NAME) or "").strip() def is_quire_configured() -> bool: """True when Quire is connected AND a default destination project is set — the precondition for the entry button, On-Star, and Automation rules.""" return bool(is_quire_connected() and quire_project_oid()) def get_quire_user_token() -> str: """Return a valid Quire access token, refreshing if expired. Empty string if the user hasn't connected their Quire account.""" with get_meta_connection() as conn: access = get_setting(conn, SETTING_QUIRE_ACCESS_TOKEN) or "" refresh = get_setting(conn, SETTING_QUIRE_REFRESH_TOKEN) or "" try: expires_at = float(get_setting(conn, SETTING_QUIRE_TOKEN_EXPIRES_AT) or 0) except TypeError, ValueError: expires_at = 0.0 if not access: return "" if time.time() < expires_at - 60: return access if not refresh: return access cid, secret = get_quire_credentials() try: data = quire_service.refresh_access_token(cid, secret, refresh) except Exception as exc: # noqa: BLE001 LOGGER.warning("[quire] token refresh failed; reconnect required: %s", exc) return "" with get_meta_connection() as conn: set_setting(conn, SETTING_QUIRE_ACCESS_TOKEN, data["access_token"]) if data.get("refresh_token"): set_setting(conn, SETTING_QUIRE_REFRESH_TOKEN, data["refresh_token"]) set_setting(conn, SETTING_QUIRE_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600)))) return data["access_token"] def quire_rate_caps() -> tuple[int, int]: """(per-minute, per-hour) caps Lectio meters its Quire calls against.""" def _cap(key: str, default: int) -> int: try: return max(1, int(get_runtime_setting(key, str(default)) or default)) except TypeError, ValueError: return default return (_cap(SETTING_QUIRE_RATE_CAP_MIN, QUIRE_RATE_DEFAULT_CAP_MIN), _cap(SETTING_QUIRE_RATE_CAP_HOUR, QUIRE_RATE_DEFAULT_CAP_HOUR)) def record_quire_call(calls: int = 1) -> None: """Log ``calls`` Quire API calls into the sliding-window meter (wired as the service usage sink at startup). Best-effort; prunes rows older than an hour.""" if calls <= 0: return try: now = int(time.time()) with get_meta_connection() as conn: conn.executemany( "INSERT INTO quire_call_log (ts) VALUES (?)", [(now,) for _ in range(int(calls))], ) conn.execute("DELETE FROM quire_call_log WHERE ts < ?", (now - 3600,)) except Exception: LOGGER.debug("[quire] failed to record %d calls", calls, exc_info=True) def get_quire_usage_status() -> dict: """Usage-meter payload: calls in the last minute/hour vs caps, plus a state flag.""" cap_min, cap_hour = quire_rate_caps() try: now = int(time.time()) with get_meta_connection() as conn: minute_used = conn.execute("SELECT COUNT(*) AS n FROM quire_call_log WHERE ts >= ?", (now - 60,)).fetchone()["n"] hour_used = conn.execute("SELECT COUNT(*) AS n FROM quire_call_log WHERE ts >= ?", (now - 3600,)).fetchone()["n"] except Exception: minute_used = hour_used = 0 minute_used, hour_used = int(minute_used), int(hour_used) blocked = minute_used >= cap_min or hour_used >= cap_hour low = (minute_used >= cap_min * 0.8) or (hour_used >= cap_hour * 0.8) state = "blocked" if blocked else ("low" if low else "ok") return {"minute_used": minute_used, "minute_cap": cap_min, "hour_used": hour_used, "hour_cap": cap_hour, "state": state} def detect_quire_plan_and_caps() -> str: """Detect the destination project's organization plan and align the rate-meter caps to it (Free 50/200, Professional 300/1250, Premium 1000/5000). Stores the plan name for display. Best-effort; returns the detected plan ("" on failure). Quire rate-limits per organization, so the cap that matters is the plan of the org owning the chosen project.""" project_oid = quire_project_oid() if not project_oid: return "" token = get_quire_user_token() if not token: return "" try: plan = quire_service.get_project_plan(token, project_oid) except Exception as exc: # noqa: BLE001 LOGGER.warning("[quire] plan detection failed: %s", exc) return "" caps = quire_service.PLAN_RATE_CAPS.get(plan.strip().lower()) with get_meta_connection() as conn: set_setting(conn, SETTING_QUIRE_PLAN, plan) if caps: # Enterprise (and any unknown plan) keeps the existing/default caps since # its quota scales with member count and isn't a fixed pair. set_setting(conn, SETTING_QUIRE_RATE_CAP_MIN, str(caps[0])) set_setting(conn, SETTING_QUIRE_RATE_CAP_HOUR, str(caps[1])) return plan def _deviantart_folder_name() -> str: return get_runtime_setting(SETTING_DEVIANTART_FOLDER_NAME) or "DeviantArt" def _apply_deviantart_image_strategy(conn: sqlite3.Connection, file_url: str) -> None: """Pin a DeviantArt feed to the 'inline' image strategy. DA feeds embed the authoritative full-size image inline (a stable, non-expiring wixmp URL), so deriving both the article lead image and the list thumbnail from that content — statelessly, every render — avoids source-page scraping and the lead-image cache getting clobbered to 'no image'. Locked (manual) so auto-detect can't change it. """ try: lead_image_service.store_feed_strategy(file_url, "inline", manual=True) except Exception: LOGGER.exception("[deviantart] failed to pin inline strategy for %s", file_url) try: conn.execute("INSERT INTO feed_display_prefs (feed_url) VALUES (?) ON CONFLICT(feed_url) DO NOTHING", (file_url,)) conn.execute("UPDATE feed_display_prefs SET thumb_strategy = 'inline' WHERE feed_url = ?", (file_url,)) except Exception: LOGGER.exception("[deviantart] failed to set inline thumb_strategy for %s", file_url) def _is_da_deactivated_error(exc: Exception) -> bool: """True when a DeviantArt add failed because the artist deactivated their account — DA returns HTTP 400 with ``"Account is inactive."``. Distinct from a 404 (deleted/renamed), so we can park it for a daily reactivation re-check instead of retrying every sync.""" return "account is inactive" in str(exc).lower() def _humanize_da_add_error(exc: Exception) -> str: """Short, user-facing reason a watch-list add failed, for the Settings list. DeviantArt gallery fetches raise ``RuntimeError('gallery fetch failed for X: HTTP 404: ...')``; surface the status where we can, else a trimmed message.""" detail = " ".join(str(exc).split()) for needle, label in ( ("Account is inactive", "deactivated"), ("404", "not found"), ("403", "access denied"), ("500", "DeviantArt error"), ): if needle in detail: return label return detail[:120] or "unknown error" def _load_da_sync_detail() -> dict: """Parse the structured sync-detail setting, or an empty shell on miss/garbage.""" raw = get_runtime_setting(SETTING_DEVIANTART_SYNC_DETAIL) if raw: try: data = json.loads(raw) if isinstance(data, dict): return data except ValueError, TypeError: pass return {"failed": [], "unwatched": []} def _da_deactivated_usernames(conn: sqlite3.Connection) -> set[str]: """Lower-cased usernames of watched artists whose DA account is deactivated.""" return {str(r["username"]).lower() for r in conn.execute("SELECT username FROM deviantart_deactivated").fetchall()} def _da_deactivated_list() -> list[dict]: """Parked deactivated artists for the Settings UI (opens its own connection).""" try: with get_meta_connection() as conn: return [ {"username": str(r["username"])} for r in conn.execute("SELECT username FROM deviantart_deactivated ORDER BY username").fetchall() ] except Exception: # noqa: BLE001 — table missing mid-migration, etc. return [] def _record_da_deactivated(conn: sqlite3.Connection, username: str) -> None: """Mark an artist deactivated (idempotent; preserves first_seen_at).""" now = datetime.now(timezone.utc).isoformat() conn.execute( "INSERT INTO deviantart_deactivated (username, first_seen_at, last_checked_at)" " VALUES (?, ?, ?)" " ON CONFLICT(username) DO UPDATE SET last_checked_at = excluded.last_checked_at", (username, now, now), ) _DA_SYNC_RESUME_FALLBACK_S = 900.0 # no Retry-After header → conservative delay def _schedule_da_sync_resume(uid: str, delay_s: float, next_round: int) -> None: def _resume() -> None: try: result = sync_deviantart_watchlist(auto_resume_round=next_round) if result.get("error"): with get_meta_connection() as conn: set_setting(conn, SETTING_DEVIANTART_SYNC_STATUS, f"Sync error: {result['error']}") except Exception as exc: LOGGER.exception("[deviantart] watchlist auto-resume failed") with get_meta_connection() as conn: set_setting(conn, SETTING_DEVIANTART_SYNC_STATUS, f"Sync failed: {_humanize_da_add_error(exc)}. Click Sync to retry.") timer = threading.Timer(delay_s, _run_in_user_context, args=(uid, _resume)) timer.daemon = True # lost on restart; the daily maintenance sync catches up timer.start() def sync_deviantart_watchlist(auto_resume_round: int = 0) -> dict: """Add a gallery feed for every artist the user Watches (add-only), and pause (disable_feed) any subscribed artist newly dropped from the watch list — see the reconcile step in _sync_deviantart_watchlist_locked. Skips artists already subscribed (anywhere) so it won't duplicate existing DeviantArt feeds. When DeviantArt's rate cap interrupts the run, schedules a background continuation (honoring Retry-After) instead of waiting for a manual re-click. Returns a result summary dict. """ token = get_deviantart_user_token() if not token: return {"added": 0, "total": 0, "error": "DeviantArt account not connected."} uid = tenancy.current_user_id() with _da_sync_lock: if uid in _da_sync_active: LOGGER.info("[deviantart] watchlist sync already running for %s — skipped", uid) return {"added": 0, "total": 0, "skipped": True} _da_sync_active.add(uid) try: return _sync_deviantart_watchlist_locked(token, uid, auto_resume_round) finally: with _da_sync_lock: _da_sync_active.discard(uid) def _sync_deviantart_watchlist_locked(token: str, uid: str, auto_resume_round: int) -> dict: cid, secret = get_deviantart_credentials() try: username = get_runtime_setting(SETTING_DEVIANTART_USERNAME) or deviantart_service.whoami(token) watching = deviantart_service.list_watching(token, username) except deviantart_service.DeviantArtRateLimited as exc: return {"added": 0, "total": 0, "error": f"DeviantArt rate limit hit — try again in a few minutes. ({exc})"} except Exception as exc: # noqa: BLE001 return {"added": 0, "total": 0, "error": f"Could not read your watch list: {exc}"} folder_name = _deviantart_folder_name() with get_meta_connection() as conn: # Exclude the combined "deviantsyouwatch" Watch feed (source='watch'): its # username is a synthetic browse endpoint, not a real artist, so it must # not count as a subscribed gallery for add-dedup or the unwatched reconcile # (otherwise it's perpetually reported as "subscribed but no longer watched"). existing = { str(r["username"]).lower() for r in conn.execute("SELECT username FROM deviantart_feeds WHERE COALESCE(source, 'gallery') != 'watch'").fetchall() } # Known-deactivated artists can never be added; skip them here so the sync # stops re-probing (and re-failing) them every run. Daily maintenance # re-checks them for reactivation. deactivated = _da_deactivated_usernames(conn) folder_id = _get_or_create_folder_by_name(conn, folder_name) # The Watch feed IS the watch list — one feed carrying every watched # artist's posts. While it exists, a per-artist gallery feed is a # duplicate of what it already delivers. has_watch_feed = bool(conn.execute("SELECT 1 FROM deviantart_feeds WHERE COALESCE(source, 'gallery') = 'watch' LIMIT 1").fetchone()) to_add = [a for a in watching if a.lower() not in existing and a.lower() not in deactivated] if has_watch_feed and to_add: # This sync and the gallery consolidation were undoing each other. The # consolidation folds a gallery feed into the Watch feed and then DELETES # its deviantart_feeds row (it must — otherwise the refresh loop # regenerates the subscription it just merged). That row is this sync's # only record of "already subscribed", so every consolidated artist came # back looking new and was re-created the next night: 537 gallery feeds # returned over two runs after the 2026-08-12 consolidation, filed back # into the folder, carrying entries the Watch feed had already delivered. LOGGER.info( "[deviantart] watchlist sync: Watch feed present — skipping %d per-artist gallery feed(s); the Watch feed already covers them", len(to_add), ) to_add = [] LOGGER.info("[deviantart] watchlist sync: %d watched, %d to add into %r", len(watching), len(to_add), folder_name) added = 0 failed = 0 deactivated_this_run = 0 failed_artists: list[dict[str, str]] = [] rate_limited = False retry_after_s: float | None = None for i, artist in enumerate(to_add, 1): if i > 1: time.sleep(0.4) # pace requests to stay under DeviantArt's rate limit try: with get_meta_connection() as conn: with get_reader() as reader: _fid, file_url = deviantart_service.create_deviantart_feed( conn, reader, artist, cid, secret, access_token=token, limit=24 ) conn.execute( "INSERT OR IGNORE INTO folder_feeds (folder_id, feed_url) VALUES (?, ?)", (folder_id, file_url), ) added += 1 except deviantart_service.DeviantArtRateLimited as exc: # Quota exhausted — stop now (the rest are re-runnable; sync is add-only # and dedupes, so the auto-resume continuation picks up where this left off). rate_limited = True retry_after_s = exc.retry_after LOGGER.info("[deviantart] watchlist sync paused at %d/%d (added=%d) — rate limited", i, len(to_add), added) break except Exception as exc: # noqa: BLE001 if _is_da_deactivated_error(exc): # Deactivated account — park it (not a hard failure); daily # maintenance re-checks for reactivation. deactivated_this_run += 1 with get_meta_connection() as conn: _record_da_deactivated(conn, artist) LOGGER.info("[deviantart] watched artist %s is deactivated — parked for re-check", artist) else: failed += 1 failed_artists.append({"username": artist, "error": _humanize_da_add_error(exc)}) LOGGER.warning("[deviantart] watchlist add failed for %s: %s", artist, exc) if i % 25 == 0: LOGGER.info("[deviantart] watchlist sync progress: %d/%d (added=%d failed=%d)", i, len(to_add), added, failed) with get_meta_connection() as conn: set_setting(conn, SETTING_DEVIANTART_SYNC_STATUS, f"Syncing {i}/{len(to_add)} (added {added})") invalidate_meta_structure_cache() invalidate_unread_counts_cache() invalidate_meta_structure_cache() invalidate_unread_counts_cache() # Reconcile: surface subscribed DA artists no longer on the watch list. # Unsubscribing stays a manual decision (curated feeds may outlive a # Watch), but leaving them actively refreshing was surprising once # noticed — so a NEWLY-unwatched artist's feed(s) get paused (disable_feed, # same as Settings' own pause/disable — no data lost, nothing unsubscribed) # once, the first run that sees them drop off the watch list. Not reapplied # on later runs: if Josh re-enables one by hand (decided to keep it live # despite the unwatch), a later sync must not fight that choice by # disabling it again every time it still shows up as unwatched. watching_lower = {a.lower() for a in watching} unwatched = sorted(u for u in existing if u not in watching_lower) if watching else [] prior_unwatched_lower = {str(u.get("username") or "").strip().lower() for u in _load_da_sync_detail().get("unwatched", [])} newly_unwatched = [u for u in unwatched if u not in prior_unwatched_lower] if newly_unwatched: with get_meta_connection() as conn: placeholders = ",".join("?" for _ in newly_unwatched) rows = conn.execute( f"SELECT id, username FROM deviantart_feeds WHERE COALESCE(source, 'gallery') != 'watch'" f" AND lower(username) IN ({placeholders})", newly_unwatched, ).fetchall() for row in rows: disable_feed(deviantart_service.feed_file_url(str(row["id"])), reason="unwatched on DeviantArt") LOGGER.info( "[deviantart] %d newly-unwatched artist(s) paused (updates disabled, not unsubscribed): %s", len(rows), ", ".join(sorted(str(r["username"]) for r in rows)), ) with get_meta_connection() as conn: set_setting(conn, SETTING_DEVIANTART_UNWATCHED_DIRTY, "1") if unwatched: LOGGER.info("[deviantart] %d subscribed artist(s) no longer watched: %s", len(unwatched), ", ".join(unwatched)) # Persist the failed/unwatched artists (structured) so the Settings UI can # list them as profile links. A fresh sync (round 0) starts clean; auto-resume # continuation rounds accumulate failures from earlier rounds (each round # processes a different slice of to_add). unwatched is recomputed from full # data every round, so it's overwritten rather than merged. prior_failed: list[dict[str, str]] = [] if auto_resume_round > 0: prior_failed = _load_da_sync_detail().get("failed", []) all_failed = prior_failed + failed_artists with get_meta_connection() as conn: set_setting( conn, SETTING_DEVIANTART_SYNC_DETAIL, json.dumps({"failed": all_failed, "unwatched": [{"username": u} for u in unwatched]}) ) remaining = len(to_add) - added - failed if rate_limited: if auto_resume_round < _DA_SYNC_MAX_AUTO_RESUMES: delay_s = max(60.0, retry_after_s or _DA_SYNC_RESUME_FALLBACK_S) _schedule_da_sync_resume(uid, delay_s, auto_resume_round + 1) final = ( f"Rate limited — added {added} so far, ~{remaining} left. " f"Auto-resuming in ~{int(delay_s // 60) or 1} min " f"(round {auto_resume_round + 1}/{_DA_SYNC_MAX_AUTO_RESUMES})." ) else: final = f"Rate limited — added {added} so far, ~{remaining} left. Auto-resume limit reached; click Sync to continue." else: # to_add now excludes both already-subscribed and known-deactivated, so # derive each count from the watch list directly rather than the residual. already = sum(1 for a in watching if a.lower() in existing) deactivated_total = sum(1 for a in watching if a.lower() in deactivated) + deactivated_this_run parts = [f"Added {added} of {len(watching)} watched"] if already: parts.append(f"{already} already subscribed") if all_failed: parts.append(f"{len(all_failed)} failed") if deactivated_total: parts.append(f"{deactivated_total} deactivated") if unwatched: parts.append(f"{len(unwatched)} subscribed but no longer watched") final = ", ".join(parts) with get_meta_connection() as conn: set_setting(conn, SETTING_DEVIANTART_SYNC_STATUS, final) return { "added": added, "failed": failed, "deactivated": deactivated_this_run, "total": len(watching), "folder": folder_name, "rate_limited": rate_limited, "unwatched": unwatched, "failed_artists": all_failed, } def _deviantart_recheck_deactivated(max_checks: int = 40) -> int: """Re-probe parked deactivated artists; subscribe any that reactivated. Runs in daily maintenance. For each parked artist, a lightweight gallery fetch tells us the current state: success → reactivated, so create the feed and drop it from the deactivated table; still inactive → bump last_checked_at; rate-limited → stop early (re-check resumes next maintenance run). Capped per run to stay polite against the API quota. Returns the count reactivated.""" token = get_deviantart_user_token() if not token: return 0 cid, secret = get_deviantart_credentials() with get_meta_connection() as conn: rows = conn.execute( "SELECT username FROM deviantart_deactivated ORDER BY last_checked_at ASC LIMIT ?", (max_checks,), ).fetchall() if not rows: return 0 folder_id = None reactivated = 0 now = datetime.now(timezone.utc).isoformat() for row in rows: artist = str(row["username"]) try: with get_meta_connection() as conn: with get_reader() as reader: if folder_id is None: folder_id = _get_or_create_folder_by_name(conn, _deviantart_folder_name()) _fid, file_url = deviantart_service.create_deviantart_feed( conn, reader, artist, cid, secret, access_token=token, limit=24 ) conn.execute( "INSERT OR IGNORE INTO folder_feeds (folder_id, feed_url) VALUES (?, ?)", (folder_id, file_url), ) conn.execute("DELETE FROM deviantart_deactivated WHERE username = ?", (artist,)) reactivated += 1 LOGGER.info("[deviantart] deactivated artist %s reactivated — subscribed", artist) except deviantart_service.DeviantArtRateLimited: LOGGER.info("[deviantart] deactivated re-check hit rate limit; stopping this run") break except Exception as exc: # noqa: BLE001 if _is_da_deactivated_error(exc): with get_meta_connection() as conn: conn.execute( "UPDATE deviantart_deactivated SET last_checked_at = ? WHERE username = ?", (now, artist), ) else: # A different error (deleted/renamed) — leave it parked; not worth # promoting to a hard failure from a background re-check. LOGGER.info("[deviantart] deactivated re-check for %s errored (non-inactive): %s", artist, exc) if reactivated: invalidate_meta_structure_cache() invalidate_unread_counts_cache() return reactivated def push_galleries_to_deviantart_watchlist() -> dict: """Watch (on DeviantArt) every artist you have a gallery feed for. Add-only.""" token = get_deviantart_user_token() if not token: return {"watched": 0, "total": 0, "error": "DeviantArt account not connected."} with get_meta_connection() as conn: usernames = [str(r["username"]) for r in conn.execute("SELECT DISTINCT username FROM deviantart_feeds").fetchall()] watched = 0 errors = 0 for name in usernames: ok, _msg = deviantart_service.watch_user(token, name) if ok: watched += 1 else: errors += 1 return {"watched": watched, "total": len(usernames), "errors": errors} def get_maintenance_hour() -> int | None: val = get_instance_setting(SETTING_MAINTENANCE_HOUR, "") if val: try: h = int(val) if 0 <= h <= 23: return h except ValueError: pass return _ENV_MAINTENANCE_HOUR _busy_work_lock = threading.Lock() _busy_work: dict[str, dict[str, int]] = {} @contextlib.contextmanager def busy_work(label: str): """Mark the current tenancy user as running long background work (also usable as a decorator). Admin "Full vacuum" refuses to touch a user's databases while any of these is open; see user_busy_reasons.""" uid = tenancy.current_user_id() with _busy_work_lock: counts = _busy_work.setdefault(uid, {}) counts[label] = counts.get(label, 0) + 1 try: yield finally: with _busy_work_lock: counts = _busy_work.get(uid, {}) counts[label] = counts.get(label, 1) - 1 if counts[label] <= 0: counts.pop(label, None) if not counts: _busy_work.pop(uid, None) def user_busy_reasons(uid: str, *, include_activity: bool = True, ignore: tuple[str, ...] = ()) -> list[str]: """Why a full vacuum of *uid*'s databases should wait, as short phrases; empty means it is safe to go ahead. Background work is read from the busy_work marks plus the existing per-user job registries (those need the user's own tenancy context). include_activity=False skips the "recently active" check, for an admin vacuuming their own account. Web, Google Reader-style and session requests touch last_seen_at; Fever and Miniflux API clients do not, so they can read as idle.""" reasons: list[str] = [] if include_activity: row = user_store.get_by_id(uid) last_seen = float(row["last_seen_at"] or 0) if row else 0.0 # last_seen_at is written at most once per throttle window, so a user active a second ago can read as up to a window old. if time.time() - last_seen < 2 * _LAST_SEEN_THROTTLE_SECONDS: reasons.append("the user was active in the last 10 minutes") with _busy_work_lock: reasons += [f"{label} is running" for label in _busy_work.get(uid, {}) if label not in ignore] with tenancy.user_context(uid): with _da_sync_lock: if uid in _da_sync_active: reasons.append("a DeviantArt sync is running") if uid in _escalation_reprobe_running: reasons.append("an escalation reprobe is running") for name, jobs in ( ("a bulk re-fetch", _refetch_jobs), ("a folder refresh", _folder_refresh_jobs), ): job = jobs.get("job") if isinstance(job, dict) and job.get("running"): reasons.append(f"{name} is running") with _autofetch_jobs_lock: # Keyed by (feed_url, entry_id), one job per post, so look through the values rather than for a single "job". if any(isinstance(j, dict) and j.get("running") for j in _autofetch_jobs.values()): reasons.append("an auto-fetch is running") return reasons def truncate_wal(conn: sqlite3.Connection, label: str, *, attempts: int = 10, delay: float = 1.0) -> bool: """Fold the WAL back and truncate it, retrying while live readers block it. ``PRAGMA wal_checkpoint(TRUNCATE)`` does not raise when a reader holds the WAL: it returns ``busy=1`` and leaves the file alone. After a VACUUM (which writes the whole rebuilt DB through the WAL) that left the WAL at ~DB size, so the total came out larger than before. Pooled connections only read briefly, so a short retry usually gets through; if not, say so instead of staying silent. """ # A busy checkpoint waits out the connection's whole busy timeout (30 s on the admin path) before it reports busy, and holds off other # writers while it waits; probe quickly instead so the retries take attempts * (0.25 s + delay) and nobody else is locked out for long. original_timeout = conn.execute("PRAGMA busy_timeout").fetchone()[0] conn.execute("PRAGMA busy_timeout=250") try: for attempt in range(attempts): busy = conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone()[0] if not busy: return True if attempt + 1 < attempts: time.sleep(delay) finally: conn.execute(f"PRAGMA busy_timeout={int(original_timeout)}") LOGGER.warning("[maintenance] %s: WAL not truncated (readers still active); it shrinks at the next checkpoint", label) return False def get_portrait_img_max_width() -> int: """Max width (px) for portrait (taller-than-wide) images in article bodies; landscape images stay full container width. 0 = disabled. Per-user DB override takes precedence over the env fallback.""" with get_meta_connection() as conn: val = get_setting(conn, SETTING_PORTRAIT_IMG_MAX_WIDTH) if val: try: w = int(val) if w >= 0: return w except ValueError: pass return _ENV_PORTRAIT_IMG_MAX_WIDTH def mark_read_on_new_tab_enabled() -> bool: """Per-user: opening a post's article in a new tab from its list title also marks it read. Off by default.""" return get_runtime_setting(SETTING_MARK_READ_ON_NEW_TAB, "0") == "1" def proxy_body_images_enabled() -> bool: """Per-user: preemptively proxy every article-pane body image through /api/img instead of loading third-party URLs directly. Off by default.""" return get_runtime_setting(SETTING_PROXY_BODY_IMAGES, "0") == "1" _PROXY_MODES = ("off", "as_needed", "always") def get_proxy_url() -> str: """Outbound-fetch proxy endpoint (e.g. gluetun's socks5h://gluetun:1080). Admin-managed; env default, DB override takes precedence.""" return get_instance_setting(SETTING_PROXY_URL, _ENV_PROXY_URL) def get_proxy_mode() -> str: """off / as_needed / always for outbound feed fetches. Per-user DB setting overrides the instance (first-admin) default; unset resolves to "off" — the proxy is opt-in everywhere.""" val = get_instance_setting(SETTING_PROXY_MODE, "off") return val if val in _PROXY_MODES else "off" def get_tailscale_url() -> str: """Last-resort outbound proxy endpoint (e.g. Tailscale's socks5h://tailscale:1080). Admin-managed; env default, DB override takes precedence. Empty means not configured, which is itself the opt-in gate — as_needed escalation never reaches this tier without one.""" return get_instance_setting(SETTING_TAILSCALE_URL, _ENV_TAILSCALE_URL) def get_flaresolverr_url() -> str: """FlareSolverr endpoint (e.g. http://flaresolverr:8191/v1). Admin-managed; env default, DB override takes precedence. Empty means not configured, same opt-in gate as the other two backends.""" return get_instance_setting(SETTING_FLARESOLVERR_URL, _ENV_FLARESOLVERR_URL) def get_img_cache_days() -> int: """Last-accessed TTL (days) for the /api/img cache. 0 = keep forever. DB override (admin-managed) takes precedence over the env fallback.""" val = get_instance_setting(SETTING_IMG_CACHE_DAYS, "") if val: try: d = int(val) if d >= 0: return d except ValueError: pass return _ENV_IMG_CACHE_DAYS def get_img_cache_max_dim() -> int: """Longest-side px to downscale cached images to. 0 = store originals as-is.""" val = get_instance_setting(SETTING_IMG_CACHE_MAX_DIM, "") if val: try: d = int(val) if d >= 0: return d except ValueError: pass return _ENV_IMG_CACHE_MAX_DIM def get_img_target_bytes() -> int: """Byte budget for a cached image, above which it is re-encoded to WebP. Separate from the max dimension because the two answer different questions: an image can sit exactly at the dimension cap and still weigh 12 MB (a 4K RGBA PNG does), and it is the bytes that make it slow to load. 0 = leave every image at whatever size it arrives. """ val = get_instance_setting(SETTING_IMG_TARGET_BYTES, "") if val: try: n = int(val) if n >= 0: return n except ValueError: pass return _ENV_IMG_TARGET_BYTES # --- Auth config --- # When set, skip the login form and auto-authenticate as the admin on every request. # Intended for local/private-network installs that don't need a password prompt. AUTO_LOGIN = os.getenv("LECTIO_AUTO_LOGIN", "0") == "1" # Password hashing scheme for stored credentials. scrypt and pbkdf2_sha256 are stdlib; # argon2 needs the optional argon2-cffi package. PASSWORD_HASH_SCHEME = os.getenv("LECTIO_PASSWORD_HASH_SCHEME", passwords.DEFAULT_SCHEME).strip().lower() if PASSWORD_HASH_SCHEME not in passwords.available_schemes(): LOGGER.warning( "LECTIO_PASSWORD_HASH_SCHEME=%r not available (have: %s); using %s.", PASSWORD_HASH_SCHEME, ", ".join(passwords.available_schemes()), passwords.DEFAULT_SCHEME, ) PASSWORD_HASH_SCHEME = passwords.DEFAULT_SCHEME # Bootstrap admin account, seeded on first startup when the users table is empty. # CHANGE the default password before exposing the instance. _DEFAULT_ADMIN_PASSWORD = "ChangeA$ap" BOOTSTRAP_ADMIN_USERNAME = os.getenv("LECTIO_ADMIN_USERNAME", "admin") BOOTSTRAP_ADMIN_PASSWORD = os.getenv("LECTIO_ADMIN_PASSWORD", _DEFAULT_ADMIN_PASSWORD) AUTH_ENABLED = True # kept as a module-level bool so tests can monkeypatch it user_store: UserStore = UserStore(AUTH_DB_PATH) SESSION_SECRET_KEY = os.getenv("LECTIO_SECRET_KEY") or secrets.token_hex(32) if AUTH_ENABLED and not os.getenv("LECTIO_SECRET_KEY"): LOGGER.warning( "LECTIO_SECRET_KEY is not set — using a random key. Sessions will not survive server restarts. Set a stable key in your .env." ) SESSION_MAX_AGE_SECONDS = 365 * 24 * 3600 # 1 year; baked into SessionMiddleware at startup # Set LECTIO_HTTPS_ONLY=1 when running behind a TLS-terminating reverse proxy. _HTTPS_ONLY = os.getenv("LECTIO_HTTPS_ONLY", "0") == "1" # Proxies trusted for X-Forwarded-* headers. "*" (default) trusts any upstream — # fine when the app port is only reachable via your reverse proxy. Set a # comma-separated allowlist (e.g. the proxy's IP) for a stricter posture. _TRUSTED_PROXIES = os.getenv("LECTIO_TRUSTED_PROXIES", "*").strip() # Have the app emit baseline security headers itself. Default off because a # reverse proxy (e.g. the bundled Traefik config) usually sets them; enable when # fronting Lectio with a proxy that does not (Caddy without a header block, a # bare/no-proxy setup, etc.). Keeps the headers from depending on the proxy. _SECURITY_HEADERS_ENABLED = os.getenv("LECTIO_SECURITY_HEADERS", "0") == "1" # Paths that are always public (no login required) _AUTH_EXEMPT_PREFIXES = ( "/login", "/static", "/healthz", "/api/img", "/api/favicon", "/api/save", "/api/bookmarklet/save", "/dev/feeds/", "/fever", "/greader/", "/v1/", "/websub/", "/sw.js", ) def _note_scheduler_progress(stage: str) -> None: """Record that the scheduled refresh advanced, and what it is doing now.""" with _scheduler_state_lock: _scheduler_state["last_progress_at"] = time.monotonic() _scheduler_state["stage"] = stage def _scheduler_stall_seconds() -> float | None: """Seconds since the in-flight pass last advanced, or None when idle. None while no pass is running: between passes there is nothing to stall. """ with _scheduler_state_lock: started = _scheduler_state.get("pass_started_at") last = _scheduler_state.get("last_progress_at") if started is None or last is None: return None return time.monotonic() - float(last) # ty: ignore[invalid-argument-type] def invalidate_meta_structure_cache() -> None: with _meta_structure_lock: _meta_structure_cache.clear() # Refresh-contention investigation (Plan.md Tier 1): this cache is designed # to invalidate only on explicit user mutations (subscribe/unsubscribe, # folder changes, feed disable/enable) -- a 2026-09-03 live capture showed # it going cold roughly every 30-45s during ordinary browsing (no such # action taken), each miss costing 5+ real seconds of get_meta_structure_snapshot's # 5-query rebuild under refresh contention. There are 37 call sites; rather # than audit them all, log the caller so the next occurrence says which one # actually fired during a routine refresh pass. try: caller = sys._getframe(1) LOGGER.info( "[perf] invalidate_meta_structure_cache from %s:%d (%s)", caller.f_code.co_filename, caller.f_lineno, caller.f_code.co_name ) except Exception: pass def invalidate_problematic_feeds_cache() -> None: with _problematic_feeds_cache_lock: _problematic_feeds_cache.clear() def get_problematic_feeds_cached(conn: sqlite3.Connection, limit: int = PROBLEMATIC_FEEDS_LIMIT) -> list[dict]: """TTL-cached problematic-feeds list (see cache comment above).""" now_pf = time.time() with _problematic_feeds_cache_lock: cached_pf = _problematic_feeds_cache.get(limit) if cached_pf and now_pf - cached_pf[0] < PROBLEMATIC_FEEDS_CACHE_TTL_SECONDS: return [dict(r) for r in cached_pf[1]] problematic_feeds = feed_refresh_service.get_problematic_feeds(conn, limit=limit) with _problematic_feeds_cache_lock: _problematic_feeds_cache[limit] = (time.time(), [dict(r) for r in problematic_feeds]) return [dict(r) for r in problematic_feeds] def is_async_action_request(request: Request, expected_header: str | None = None) -> bool: requested_with = (request.headers.get("X-Requested-With") or "").strip() if not requested_with: return False if expected_header is None: return requested_with.startswith("lectio-") return requested_with == expected_header @asynccontextmanager async def lifespan(app: FastAPI): # Startup _attach_pending_access_filter() ensure_thumb_schema() ensure_img_cache_schema() ensure_yt_duration_schema() ensure_websub_schema() bootstrap_admin() # Bring every existing user's meta/starred schema up to current. Per-user DBs # are otherwise schema-init'd only at provision time, so any table added # after a user was provisioned (e.g. feed_fetch_history) is missing from # their DB and surfaces as a "no such table" 500 (Feed Properties, etc.). # ensure_*_schema are idempotent, so this is a cheap no-op once migrated. def _ensure_user_schema() -> None: ensure_meta_schema() ensure_starred_archive_schema() ensure_reader_indexes() migrate_spaced_manual_tags() _for_each_background_user("per-user schema migration", _ensure_user_schema) # Kill switch for heavy startup work. Set LECTIO_DISABLE_STARTUP_BACKFILL=1 # to skip the scheduled-refresh loop, the FTS index build, and the # lead-image / YouTube / archive / read-history backfills. Useful after an # OPML import of hundreds of feeds (boot calm, let pages render, then unset # and restart). The test suite also sets it so these background DB writers # don't race a test's own DB ops on the same temp DB ("database is locked"). backfill_disabled = os.getenv("LECTIO_DISABLE_STARTUP_BACKFILL", "0") == "1" if backfill_disabled: LOGGER.warning("LECTIO_DISABLE_STARTUP_BACKFILL=1 — skipping FTS build, scheduled refresh, and backfill threads") # reader's FTS index is no longer built or maintained: both search surfaces # resolve in SQL (_search_entry_keys_in_sql, _filter_star_keys_by_search) # because search_entries builds a highlighted snippet per result, which # measured at ~95% of a 10-20s search. Keeping the index cost 1.3ms per new # entry on every refresh plus a file about the size of the reader DB itself # (564MB against 743MB on the live library) — all of it for an index no # query touched. scripts/drop_search_index.py reclaims it on an existing # install; a fresh one never creates it. if LECTIO_PUBLIC_URL: _migrate_websub_to_shared() def _prepare_user_dbs() -> None: # Checkpoint both WAL files so the first request does not have to rebuild # the WAL index from a large file left over from the previous run. Use # direct connections so we're not racing the reader. for _ckpt_path in (tenancy.reader_db_path(), tenancy.meta_db_path()): try: with contextlib.closing(sqlite3.connect(str(_ckpt_path), timeout=5)) as _ckpt_conn: _ckpt_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") except Exception: pass with get_meta_connection() as conn: purge_lower_level_folders(conn) with _app_settings_cache_lock: _app_settings_cache[tenancy.current_user_id()] = _load_app_settings_cache(conn) with get_reader(): # creates the reader DB if missing pass _for_each_background_user("per-user DB prep", _prepare_user_dbs) app.state.last_scheduled_refresh_started_at = time.monotonic() # Warm YouTube duration in-memory cache from DB so first renders are instant. youtube_duration_service.warm_cache_from_db() # Warm lead image cache from DB so thumbnails are available on first render. # Must run per-user: lead images live in each tenant's own meta DB, and the # render path (get_cached_entry_thumbnail) only consults the in-memory cache # with no per-user DB fallback, so every user's cache must be warmed or their # thumbnails stay blank until the rate-limited background backfill catches up. _for_each_background_user("lead-image cache warm", lead_image_service.warm_cache_from_db) # Warm the per-user unread-counts cache so the first home load after a restart # doesn't pay the synchronous reader-DB scan under startup contention (cold # cache + concurrent first requests + a refresh kicking off was producing # multi-second — occasionally ~20s — home loads right after a redeploy). def _warm_unread_counts() -> None: try: counts = _compute_unread_counts_by_feed() with unread_counts_cache_lock: unread_counts_cache["unread_counts"] = (time.time(), counts) except Exception: LOGGER.debug("unread-counts cache warm failed", exc_info=True) _for_each_background_user("unread-counts cache warm", _warm_unread_counts) # Ensure existing scraped file:// feeds have their entries imported into the # reader DB. Feeds may be in backoff from stale "no retriever" errors that # pre-date the feed_root='' fix — clear that state and force a reader sync. def _sync_scraped_feeds() -> None: try: with get_meta_connection() as _sc: _sf_rows = _sc.execute("SELECT id, mode FROM scraped_feeds").fetchall() for _sf_row in _sf_rows: _sf_id = str(_sf_row["id"]) _sf_url = scraper_service.feed_file_url(_sf_id) try: with get_meta_connection() as _sc: _sc.execute("DELETE FROM feed_failure_state WHERE feed_url = ?", (_sf_url,)) # link_list feeds seed initial links as hidden so only new links # appear as entries. If ALL entries are still hidden (nothing has # appeared yet), unhide them so the feed isn't permanently empty. if str(_sf_row["mode"]) == "link_list": _visible = _sc.execute( "SELECT count(*) FROM scraped_entries WHERE scraped_feed_id=? AND NOT hidden", (_sf_id,), ).fetchone()[0] if _visible == 0: _sc.execute( "UPDATE scraped_entries SET hidden=0 WHERE scraped_feed_id=?", (_sf_id,), ) scraper_service._write_feed_file(_sc, _sf_id) with get_reader() as _r: _r.update_feed(_sf_url) except Exception: pass except Exception: LOGGER.exception("[scraper] startup scraped-feed sync failed") if not backfill_disabled: threading.Thread( target=lambda: _for_each_background_user("scraped-feed sync", _sync_scraped_feeds), daemon=True, name="sync-scraped-feeds", ).start() # Auto-taggers and the dedup cleanup write per-user tag/strategy state, so # run them once per background user (each under its own tenancy context). def _startup_auto_tag_and_dedup() -> None: # Artwork tagger runs first so webcomic tagger won't clobber ArtStation # feeds that live in folders whose name also contains "comic". _auto_tag_artwork_feeds() # Auto-tag feeds in "comic*" folders with strategy='webcomic'. _auto_tag_webcomic_feeds() # GitHub release feeds get og_scrape + no list thumbnail. _auto_tag_github_release_feeds() # Retroactive dedup: suppress duplicate unread entries (same slug or # title+date) that accumulated before _suppress_guid_churn could catch them. try: with get_reader() as reader: with get_meta_connection() as conn: cleaned = _cleanup_intra_feed_slug_dupes(reader, conn) if cleaned: LOGGER.info("[guid-churn-cleanup] suppressed %d pre-existing duplicate entries", cleaned) except Exception: LOGGER.exception("[guid-churn-cleanup] startup cleanup failed") _for_each_background_user("auto-tag/dedup", _startup_auto_tag_and_dedup) stop_event = threading.Event() if not backfill_disabled: thread = threading.Thread( target=scheduled_refresh_loop, args=(stop_event,), daemon=True, ) app.state.refresh_stop_event = stop_event app.state.refresh_thread = thread thread.start() # Shares the refresh loop's stop_event: the watchdog exists only to watch # that loop, so the two start and stop together. watchdog = threading.Thread( target=scheduler_watchdog_loop, args=(stop_event,), daemon=True, ) app.state.scheduler_watchdog_thread = watchdog watchdog.start() else: app.state.refresh_stop_event = stop_event app.state.refresh_thread = None app.state.scheduler_watchdog_thread = None # Backfill durations for any existing YouTube entries not yet stored. def _backfill() -> None: with get_meta_connection() as conn: rows = conn.execute( "SELECT DISTINCT feed_url FROM folder_feeds WHERE feed_url LIKE '%youtube.com/feeds/videos.xml%'" ).fetchall() for row in rows: youtube_duration_service.fetch_and_store_durations_for_feed(str(row["feed_url"])) if not backfill_disabled: threading.Thread( target=lambda: _for_each_background_user("youtube-duration backfill", _backfill), daemon=True, ).start() # Backfill lead images for all feeds whose entries haven't been checked yet. # `force_retry_negative=False` keeps this incremental: only feeds that have # never been checked do work, so a startup right after an OPML import of # hundreds of feeds doesn't flood the meta DB with writes and starve the # request path. def _backfill_lead_images() -> None: with get_meta_connection() as conn: rows = conn.execute("SELECT DISTINCT feed_url FROM folder_feeds").fetchall() for row in rows: lead_image_service.fetch_and_store_lead_images_for_feed(str(row["feed_url"]), force_retry_negative=False) if not backfill_disabled: threading.Thread( target=lambda: _for_each_background_user("lead-image backfill", _backfill_lead_images), daemon=True, ).start() # Start the starred archive worker, then backfill pending rows for any # saved entries that don't yet have a complete archive (covers the # initial rollout and any re-stars after maintenance pruning). Also # one-shot metadata backfill for archive rows that completed before # title/link/etc columns were added to the schema. if not backfill_disabled: starred_archive_service.start_worker() def _archive_backfill_task() -> None: starred_archive_service.backfill_saved_entries_from_archive() starred_archive_service.backfill_missing_archives() starred_archive_service.backfill_metadata_for_complete_rows() if not backfill_disabled: threading.Thread( target=lambda: _for_each_background_user("starred-archive backfill", _archive_backfill_task), daemon=True, name="starred-archive-backfill", ).start() # One-time backfill: populate read_history from entry_read_state if history is empty. def _backfill_read_history() -> None: try: with get_meta_connection() as conn: count = conn.execute("SELECT COUNT(*) FROM read_history").fetchone()[0] if count > 0: return rows = conn.execute( "SELECT feed_url, entry_id, read_at FROM entry_read_state ORDER BY read_at DESC LIMIT ?", (READ_HISTORY_CAP,), ).fetchall() if not rows: return needed = {(str(r["feed_url"]), str(r["entry_id"])): str(r["read_at"]) for r in rows} to_insert = [] with get_reader() as reader: for (feed_url, entry_id), read_at in needed.items(): entry = reader.get_entry((feed_url, entry_id), None) if entry is None: continue feed = reader.get_feed(feed_url, None) to_insert.append( ( feed_url, entry_id, str(getattr(entry, "title", None) or ""), str(getattr(entry, "link", None) or ""), str(getattr(feed, "resolved_title", None) or getattr(feed, "title", None) or ""), read_at, ) ) if not to_insert: return with get_meta_connection() as conn: conn.executemany( "INSERT INTO read_history (feed_url, entry_id, title, link, feed_title, read_at)" " VALUES (?, ?, ?, ?, ?, ?)" " ON CONFLICT(feed_url, entry_id) DO NOTHING", to_insert, ) LOGGER.info("[read_history] backfilled %d entries from entry_read_state", len(to_insert)) except Exception: LOGGER.exception("[read_history] backfill error") if not backfill_disabled: threading.Thread( target=lambda: _for_each_background_user("read-history backfill", _backfill_read_history), daemon=True, name="read-history-backfill", ).start() # Daily Maintenance loop — runs once per day at the configured maintenance hour. # Gated by the same kill switch as the other daemons: the loop is a persistent # thread that ticks every 30s doing per-user DB writes (email-batch flush, and # a full maintenance run when due). Under the test suite (which sets # LECTIO_DISABLE_STARTUP_BACKFILL=1) it would otherwise keep running for the # rest of the pytest process once any TestClient test triggered the lifespan, # racing later tests' fresh-DB setup as an intermittent "database is locked". # Tests that need maintenance behavior call _run_daily_maintenance() directly. maint_stop_event = threading.Event() app.state.maint_stop_event = maint_stop_event if not backfill_disabled: threading.Thread( target=_daily_maintenance_loop, args=(maint_stop_event,), daemon=True, name="daily-maintenance", ).start() # In debug mode, auto-subscribe dev feeds to the _Lectio folder. if DEBUG_MODE: _dev_base = "http://127.0.0.1:8000" _dev_urls = [ f"{_dev_base}/dev/feeds/email-match.xml", f"{_dev_base}/dev/feeds/email-match.atom", f"{_dev_base}/dev/feeds/email-match.json", f"{_dev_base}/dev/feeds/email-skip.xml", f"{_dev_base}/dev/feeds/email-skip.atom", f"{_dev_base}/dev/feeds/email-skip.json", ] with get_meta_connection() as conn: _lectio_folder_id = _get_lectio_folder_id(conn) if _lectio_folder_id: _dev_newly_added: list[str] = [] for _dev_url in _dev_urls: try: with get_meta_connection() as _c: _already = _c.execute( "SELECT 1 FROM folder_feeds WHERE folder_id = ? AND feed_url = ?", (_lectio_folder_id, _dev_url), ).fetchone() if not _already: add_feed_to_folder(_dev_url, _lectio_folder_id) _dev_newly_added.append(_dev_url) except Exception: LOGGER.exception("[dev-feeds] failed to subscribe %s", _dev_url) if _dev_newly_added: feed_refresh_service.update_feeds(_dev_newly_added) try: yield finally: stop_event = getattr(app.state, "refresh_stop_event", None) thread = getattr(app.state, "refresh_thread", None) if stop_event and thread: stop_event.set() # A scheduled refresh in flight can take tens of seconds (one HTTP # timeout per feed). Wait long enough to let it finish cleanly so # we don't leave the meta DB in a partial-write state. shutdown_timeout = float(os.getenv("LECTIO_SHUTDOWN_TIMEOUT_SECONDS", "30")) thread.join(timeout=shutdown_timeout) if thread.is_alive(): LOGGER.warning( "[shutdown] refresh worker did not finish within %.0fs; abandoning (daemon thread will be killed by interpreter exit)", shutdown_timeout, ) # Stop the starred archive worker — short timeout, since the only # work in flight is HTTP fetches with 15s ceilings. try: starred_archive_service.stop_worker(timeout=5.0) except Exception as exc: # noqa: BLE001 LOGGER.warning("[shutdown] starred archive worker stop failed: %s", exc) yt_stop = getattr(app.state, "yt_stop_event", None) if yt_stop: yt_stop.set() app = FastAPI(title="Lectio", lifespan=lifespan) def _session_logged_in(request: Request) -> bool: """Whether the session is fully authenticated (requires a valid user_id).""" if not request.session.get("authenticated"): return False uid = request.session.get("user_id") return bool(uid and tenancy.is_valid_user_id(uid)) class _AuthMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): if not AUTH_ENABLED: return await call_next(request) path = request.url.path if any(path.startswith(p) for p in _AUTH_EXEMPT_PREFIXES): return await call_next(request) if _session_logged_in(request): return await call_next(request) if AUTO_LOGIN: # Auto-authenticate as the first non-disabled admin user without # showing the login form. Intended for local/private-network installs. for _u in user_store.list_users(): if _u.get("is_admin") and not _u.get("disabled"): request.session["authenticated"] = True request.session["user_id"] = _u["user_id"] break if _session_logged_in(request): return await call_next(request) # Store a same-origin relative path (path + query) so _safe_next accepts # it and the user lands back on the deep link after logging in — e.g. a # /?subscribe= quick-subscription link. An absolute URL would be # rejected by _safe_next as an open-redirect and drop the user on "/". next_url = request.url.path if request.url.query: next_url += "?" + request.url.query return RedirectResponse(url=f"/login?next={quote_plus(next_url)}", status_code=303) class _SecurityHeadersMiddleware: """Emit baseline security response headers from the app (opt-in via LECTIO_SECURITY_HEADERS=1), so they hold regardless of the reverse proxy. Mirrors the headers the bundled Traefik config sets. Existing headers are not overwritten, so a proxy that already sets them wins and there are no duplicates. HSTS is only added when LECTIO_HTTPS_ONLY=1.""" def __init__(self, app, *, hsts: bool): self.app = app self._headers: list[tuple[bytes, bytes]] = [ (b"x-content-type-options", b"nosniff"), (b"x-frame-options", b"DENY"), (b"referrer-policy", b"no-referrer-when-downgrade"), ] if hsts: self._headers.append((b"strict-transport-security", b"max-age=31536000; includeSubDomains")) async def __call__(self, scope, receive, send): if scope.get("type") != "http": return await self.app(scope, receive, send) async def send_wrapper(message): if message["type"] == "http.response.start": headers = message.setdefault("headers", []) present = {k.lower() for k, _ in headers} for key, val in self._headers: if key not in present: headers.append((key, val)) await send(message) await self.app(scope, receive, send_wrapper) def _touch_user_last_seen(uid: str) -> None: """Record per-user activity time, throttled (this runs on every request).""" now = time.time() with _last_seen_touch_lock: if now - _last_seen_touch.get(uid, 0.0) < _LAST_SEEN_THROTTLE_SECONDS: return _last_seen_touch[uid] = now try: user_store.touch_last_seen(uid, now) except Exception: LOGGER.debug("touch_last_seen failed", exc_info=True) class _TenancyMiddleware: """Bind the request's user into the tenancy context. Pure-ASGI and registered innermost (downstream of the auth gate), so the binding wraps the route handler. Sync handlers run via anyio's threadpool, which copies the current contextvars into the worker thread, so a value set here is visible to get_reader() / get_meta_connection() deep in the call stack. Requests without an authenticated session user or API token (static assets, the login page, other unauthenticated hits) run unbound: anything they touch must be global, and a per-user DB access raises TenancyUnboundError. """ def __init__(self, app): self.app = app @staticmethod def _greader_user_from_scope(scope) -> str | None: """Resolve a /greader/ request's bearer token to a user, using only the Authorization header / query string (no body read needed).""" token = "" for key, val in scope.get("headers", []): if key == b"authorization": auth = val.decode("latin-1") if auth.startswith("GoogleLogin auth="): token = auth[17:].strip() elif auth.startswith("Bearer "): token = auth[7:].strip() break if not token: qs = parse_qs(scope.get("query_string", b"").decode("latin-1")) token = (qs.get("token") or [""])[0] return user_store.resolve_greader_token(token) async def __call__(self, scope, receive, send): if scope.get("type") != "http": return await self.app(scope, receive, send) path = scope.get("path", "") uid: str | None = None if path.startswith("/greader/"): # API requests carry their identity in a bearer token, not a session. uid = self._greader_user_from_scope(scope) else: session = scope.get("session") or {} sid = session.get("user_id") if sid and session.get("authenticated") and tenancy.is_valid_user_id(sid): uid = sid if uid and tenancy.is_valid_user_id(uid): _touch_user_last_seen(uid) token = tenancy.set_current_user(uid) try: return await self.app(scope, receive, send) finally: tenancy.reset_current_user(token) return await self.app(scope, receive, send) # Paths exempt from CSRF validation. /login is the auth gate itself (rate- # limited separately). /static and /healthz are GET-only anyway, but listing # explicitly documents intent. _CSRF_EXEMPT_PREFIXES = ( "/login", "/static", "/healthz", "/api/img", "/api/favicon", "/api/save", "/api/bookmarklet/save", "/fever", "/greader/", "/v1/", "/websub/", ) _CSRF_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) _CSRF_SESSION_KEY = "csrf_token" _CSRF_HEADER_NAME = "x-csrf-token" _CSRF_FORM_FIELD = "_csrf" def _ensure_csrf_token(session: dict) -> str: """Return the session's CSRF token, generating one on first use. Token is per-session (rotated whenever the session itself rotates, e.g. on logout/login). Stored in the session cookie which is HttpOnly + signed so attackers on other origins can neither read nor forge it. """ token = session.get(_CSRF_SESSION_KEY) if not token: token = secrets.token_urlsafe(32) session[_CSRF_SESSION_KEY] = token return token class _AccessLogMiddleware: """Replaces uvicorn's built-in access log so we can suppress the prefetch 204/503s without wrestling with uvicorn's logger config (filters added to uvicorn.access weren't taking effect in this deployment). Format mirrors uvicorn's: ``CLIENT - "METHOD PATH HTTP/x.y" STATUS REASON``. """ _SUPPRESS_STATUSES = (204, 503) _STATUS_REASONS = { 200: "OK", 204: "No Content", 301: "Moved Permanently", 302: "Found", 303: "See Other", 304: "Not Modified", 307: "Temporary Redirect", 308: "Permanent Redirect", 400: "Bad Request", 401: "Unauthorized", 403: "Forbidden", 404: "Not Found", 405: "Method Not Allowed", 409: "Conflict", 413: "Payload Too Large", 429: "Too Many Requests", 500: "Internal Server Error", 502: "Bad Gateway", 503: "Service Unavailable", } def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope["type"] != "http": await self.app(scope, receive, send) return _t0 = time.monotonic() status_holder = [0] async def send_wrapper(message): if message.get("type") == "http.response.start": status_holder[0] = int(message.get("status", 0)) await send(message) try: await self.app(scope, receive, send_wrapper) finally: status = status_holder[0] elapsed_ms = int((time.monotonic() - _t0) * 1000) client = scope.get("client") or ("?", 0) client_addr = f"{client[0]}:{client[1]}" method = scope.get("method", "?") raw_path = scope.get("raw_path", b"") or scope.get("path", "").encode() qs = scope.get("query_string", b"") or b"" request_path = raw_path.decode("latin-1", errors="replace") if qs: request_path = f"{request_path}?{qs.decode('latin-1', errors='replace')}" http_version = scope.get("http_version", "1.1") request_line = f"{method} {request_path} HTTP/{http_version}" # Suppression: prefetch 204/503 with list_feed_url= are pure noise. if not (status in self._SUPPRESS_STATUSES and "list_feed_url=" in request_path): reason = self._STATUS_REASONS.get(status, "") duration = f" {elapsed_ms}ms" if elapsed_ms >= 200 else "" LOGGER.info('%s - "%s" %d %s%s', client_addr, request_line, status, reason, duration) class _RejectPrefetchMiddleware: """Two-layer defense against browser link-prefetch floods. 1. **Drop prefetches outright.** Browsers tag them with one of several headers (Sec-Purpose, Purpose, X-Moz, Sec-Fetch-Dest=empty for nav-style prefetch). Match generously — false positives just make a prefetch miss, which is fine. 2. **Concurrency cap on home() requests.** Backstop in case detection misses a browser variant: hold a semaphore so only N home renders run at once. Excess returns 503, which tells the prefetcher to stop. /healthz and /static bypass both layers so probes and asset prefetches still work. """ PASSTHROUGH_PREFIXES = ("/healthz", "/static", "/login", "/api/img", "/api/favicon") def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope["type"] != "http": await self.app(scope, receive, send) return path = scope.get("path", "") if any(path.startswith(p) for p in self.PASSTHROUGH_PREFIXES): await self.app(scope, receive, send) return headers = scope.get("headers") or [] is_prefetch = False is_spa_fetch = False sec_fetch_dest = b"" sec_fetch_mode = b"" x_requested_with = b"" user_agent = b"" referer = b"" for header_name, header_value in headers: name = header_name.decode("latin-1").lower() if name == "sec-purpose": if b"prefetch" in header_value.lower() or b"prerender" in header_value.lower(): is_prefetch = True elif name == "purpose": low = header_value.lower() if b"prefetch" in low or b"preview" in low or b"prerender" in low: is_prefetch = True elif name == "x-moz": if b"prefetch" in header_value.lower() or b"prerender" in header_value.lower(): is_prefetch = True elif name == "sec-fetch-dest": sec_fetch_dest = header_value elif name == "sec-fetch-mode": sec_fetch_mode = header_value elif name == "x-requested-with": x_requested_with = header_value if b"lectio" in header_value.lower(): is_spa_fetch = True elif name == "user-agent": user_agent = header_value elif name == "referer": referer = header_value # Heuristic: browser-side prefetch via fetch() (Chrome NoState Prefetch # and similar) doesn't always set Sec-Purpose, but it has a unique # fingerprint — Sec-Fetch-Dest=empty + Sec-Fetch-Mode=cors on a path # that isn't an actual API endpoint, with no X-Requested-With from our # SPA. Real SPA fetches always send X-Requested-With=lectio-...; real # top-level navigations send Sec-Fetch-Dest=document Sec-Fetch-Mode=navigate. if not is_prefetch and not is_spa_fetch and sec_fetch_dest == b"empty" and sec_fetch_mode == b"cors" and path == "/": is_prefetch = True # Diagnostic: dump the first few suspect-looking requests so we can # see what's actually firing them. Capped to avoid log spam. with _prefetch_header_log_lock: remaining = _prefetch_header_log_remaining[0] if remaining > 0: with _prefetch_header_log_lock: if _prefetch_header_log_remaining[0] > 0: _prefetch_header_log_remaining[0] -= 1 LOGGER.info( "[prefetch-diag] path=%s qs=%r dest=%r mode=%r xrw=%r referer=%r ua=%r", path, scope.get("query_string", b"")[:200], sec_fetch_dest, sec_fetch_mode, x_requested_with, referer[:120], user_agent[:120], ) if is_prefetch: response = Response(status_code=204, headers={"Cache-Control": "no-store"}) await response(scope, receive, send) return if path == "/" or path.startswith("/?"): await self.app(scope, receive, send) return await self.app(scope, receive, send) class _CSRFMiddleware: """Pure-ASGI CSRF protection. On unsafe methods (POST/PUT/PATCH/DELETE) requires a token that matches the session's stored token. Token is supplied either via: - X-CSRF-Token header (used by SPA fetch handlers), or - `_csrf` form field (for plain HTML form submits / multipart uploads). Read-only methods are allowed through unchanged. The body is buffered and re-streamed to downstream handlers via a replay receive callable so route handlers can still parse it normally. """ def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope["type"] != "http": await self.app(scope, receive, send) return # Ensure every request has a CSRF token in its session so templates # can render it and SPA fetches can read it from a meta tag. The # token is created lazily on first request after a fresh session. session = scope.get("session") if session is not None and not session.get(_CSRF_SESSION_KEY): session[_CSRF_SESSION_KEY] = secrets.token_urlsafe(32) method = scope.get("method", "GET").upper() path = scope.get("path", "") if method in _CSRF_SAFE_METHODS or any(path.startswith(p) for p in _CSRF_EXEMPT_PREFIXES): await self.app(scope, receive, send) return # Buffer the request body so downstream handlers can still read it. body_chunks: list[bytes] = [] more_body = True while more_body: message = await receive() if message["type"] != "http.request": # Unexpected (e.g. http.disconnect); pass through unchanged. await self.app(scope, receive, send) return body_chunks.append(message.get("body", b"")) more_body = bool(message.get("more_body", False)) body = b"".join(body_chunks) # Token from header (preferred for SPA fetches). submitted: str | None = None for header_name, header_value in scope.get("headers") or []: if header_name.decode("latin-1").lower() == _CSRF_HEADER_NAME: submitted = header_value.decode("latin-1") break # Fall back to a `_csrf` field in form-encoded bodies. if not submitted and body: content_type = b"" for header_name, header_value in scope.get("headers") or []: if header_name.decode("latin-1").lower() == "content-type": content_type = header_value break ctype = content_type.decode("latin-1", errors="ignore").lower() if ctype.startswith("application/x-www-form-urlencoded"): from urllib.parse import parse_qs try: form = parse_qs(body.decode("utf-8", errors="ignore")) values = form.get(_CSRF_FORM_FIELD, []) if values: submitted = values[0] except Exception: submitted = None elif ctype.startswith("multipart/form-data"): # Cheap multipart probe: scan the raw body for the field name. # Avoids pulling in a streaming parser for the one OPML form. marker = b'name="' + _CSRF_FORM_FIELD.encode() + b'"' idx = body.find(marker) if idx >= 0: # The value sits two CRLFs after the disposition line. sep = b"\r\n\r\n" val_start = body.find(sep, idx) if val_start >= 0: val_start += len(sep) val_end = body.find(b"\r\n", val_start) if val_end >= 0: try: submitted = body[val_start:val_end].decode("utf-8", errors="ignore") except Exception: submitted = None expected = (scope.get("session") or {}).get(_CSRF_SESSION_KEY) if not expected or not submitted or not secrets.compare_digest(str(submitted), str(expected)): response = JSONResponse( {"detail": "CSRF token missing or invalid."}, status_code=403, ) await response(scope, receive, send) return # Replay the buffered body to downstream handlers. replayed = False async def replay_receive(): nonlocal replayed if not replayed: replayed = True return {"type": "http.request", "body": body, "more_body": False} return {"type": "http.disconnect"} await self.app(scope, replay_receive, send) # Middleware order matters. Starlette's add_middleware does insert(0) and the # stack-builder reverses, so LAST add_middleware = OUTERMOST = runs FIRST # inbound. We need this inbound chain: # _RejectPrefetchMiddleware (drop browser speculation-rules prefetches) # ProxyHeaders (rewrite client IP / scheme from X-Forwarded-*) # → SessionMiddleware (parse signed session cookie → scope["session"]) # → _CSRFMiddleware (validate POST tokens; needs scope["session"]) # → _AuthMiddleware (gate on session["authenticated"]; needs scope["session"]) # → app # Therefore add Auth FIRST (innermost) and outer middlewares LAST. # _TenancyMiddleware is added before Auth so it is innermost of all — it binds # the per-user context right around the endpoint, after the auth gate has run. app.add_middleware(_TenancyMiddleware) app.add_middleware(_AuthMiddleware) app.add_middleware(_CSRFMiddleware) app.add_middleware( SessionMiddleware, secret_key=SESSION_SECRET_KEY, max_age=SESSION_MAX_AGE_SECONDS, https_only=_HTTPS_ONLY, same_site="lax", ) # When running behind Traefik/nginx/Caddy, trust forwarded headers so # uvicorn sees the real scheme and client IP. if _HTTPS_ONLY: from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware _trusted_hosts = "*" if _TRUSTED_PROXIES == "*" else [h.strip() for h in _TRUSTED_PROXIES.split(",") if h.strip()] app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=_trusted_hosts) # Drop prefetch traffic before any other middleware does work. app.add_middleware(_RejectPrefetchMiddleware) # App-emitted security headers (opt-in), outermost so they apply to every # response regardless of which inner middleware produced it. if _SECURITY_HEADERS_ENABLED: app.add_middleware(_SecurityHeadersMiddleware, hsts=_HTTPS_ONLY) # Compress responses before they reach Traefik so Traefik's own compress # middleware sees Content-Encoding: gzip and passes through without buffering. # Buffering the full 500KB page before sending a single byte was the main # culprit for slow first-byte times on variable connections. app.add_middleware(GZipMiddleware, minimum_size=1000) # Access log replaces uvicorn's built-in (disabled via --no-access-log) so we # can suppress prefetch 204/503 noise. Add LAST so it sees the final response. app.add_middleware(_AccessLogMiddleware) class _CachedStaticFiles(StaticFiles): """StaticFiles that sends a long-lived Cache-Control header. Safe because every /" ) if re.search(r"", sanitized, re.IGNORECASE): sanitized = re.sub(r"()", picker_assets + r"\1", sanitized, count=1, flags=re.IGNORECASE) else: sanitized += picker_assets return HTMLResponse(sanitized, status_code=200) def is_probably_frame_blocked(headers: httpx.Headers) -> tuple[bool, str]: xfo = (headers.get("x-frame-options") or "").lower() if "deny" in xfo: return True, "x-frame-options=deny" if "sameorigin" in xfo: return True, "x-frame-options=sameorigin" csp = (headers.get("content-security-policy") or "").lower() if "frame-ancestors" in csp: match = re.search(r"frame-ancestors\s+([^;]+)", csp) if match: value = match.group(1).strip() if "*" not in value: return True, f"csp frame-ancestors={value}" return False, "" def probe_frameability(source_url: str) -> dict[str, object]: parsed = urlparse(source_url) if parsed.scheme not in {"http", "https"}: return {"blocked": True, "reason": "unsupported-url-scheme"} try: source_url = url_guard.ensure_safe_outbound_url(source_url) with url_guard.build_client(timeout=8.0, headers={"User-Agent": READABILITY_USER_AGENT}) as client: with client.stream("GET", source_url) as response: blocked, reason = is_probably_frame_blocked(response.headers) final_url = str(response.url) return { "blocked": blocked, "reason": reason, "final_url": final_url, } except Exception as exc: # If probing fails, do not force-block; UI can still attempt direct load then fallback. return { "blocked": False, "reason": f"probe-failed: {exc}", "final_url": source_url, } # Comment threads are never article content, but they defeat extraction: a # blogspot/WordPress comments section carries hundreds of avatar/delete-icon # images and huge text, so readability scores it above the actual post and the # "more images = better content" fallback can't rescue it (the comic post-body # has *fewer* images than the comment avatars). Stripping the section before # extraction lets the real body win. Container-level selectors only — never a # broad `[class*=comment]`, which would catch content like "commentary" or a # "N comments" badge. _COMMENT_SECTION_SELECTORS = ( "#comments", ".comments", ".comments-area", ".comment-list", ".comment-thread", ".comment-section", "#comment-holder", "#disqus_thread", ".disqus", "#respond", ".comment-form", "#comment-form", ) def _strip_comment_sections(raw_html: str) -> str: """Remove comment-thread containers from page HTML before extraction.""" try: from bs4 import BeautifulSoup soup = BeautifulSoup(raw_html, "html.parser") removed = False for sel in _COMMENT_SECTION_SELECTORS: for el in soup.select(sel): el.decompose() removed = True return str(soup) if removed else raw_html except Exception: # noqa: BLE001 return raw_html # In-body chrome that modern CMSes embed *inside* the article container, so it # survives a content-selector match (Future plc's #article-body carries the # social-share bar, follow/newsletter widgets, ad units and a related-posts # aside among the tab figures). Removed before extraction so neither readability # nor the selector/whole-body fallbacks drag it in. Kept to markers that are # never article prose: named share/newsletter/ad/affiliate widgets, tooltips, # and