Typecheck enforcement

pyproject.toml

@@ -105,7 +105,7 @@ ignore = ["D", "EM", "FBT", "PLR2004", "UP035", "TRY003", "E501", "TD002", "S104
 include = ["shelfmark"]
 exclude = [".local", "tests", "**/__pycache__", "**/node_modules"]
 pythonVersion = "3.14"
-typeCheckingMode = "off"
+typeCheckingMode = "standard"
 
 [tool.vulture]
 paths = ["shelfmark"]

shelfmark/__main__.py

@@ -4,5 +4,20 @@
 from shelfmark.core.config import config
 from shelfmark.main import app, socketio
 
+
+def _resolve_debug_flag(value: object) -> bool:
+    """Normalize DEBUG config values for Flask-SocketIO startup."""
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() in {"1", "true", "yes", "on"}
+    return bool(value)
+
+
 if __name__ == "__main__":
-    socketio.run(app, host=FLASK_HOST, port=FLASK_PORT, debug=config.get("DEBUG", False))
+    socketio.run(
+        app,
+        host=FLASK_HOST,
+        port=FLASK_PORT,
+        debug=_resolve_debug_flag(config.get("DEBUG", False)),
+    )

shelfmark/api/websocket.py

@@ -56,6 +56,11 @@ def is_enabled(self) -> bool:
         """Check if WebSocket is enabled and ready."""
         return self._enabled and self.socketio is not None
 
+    def _get_socketio(self) -> SocketIO | None:
+        if not self._enabled:
+            return None
+        return self.socketio
+
     def set_queue_status_fn(self, fn: Callable) -> None:
         """Set the queue_status function reference for per-room filtering."""
         self._queue_status_fn = fn
@@ -126,12 +131,13 @@ def leave_user_room(
 
     def broadcast_status_update(self, status_data: dict[str, Any]) -> None:
         """Broadcast status update to all connected clients, filtered by user room."""
-        if not self.is_enabled():
+        socketio = self._get_socketio()
+        if socketio is None:
             return
 
         try:
             # Admins (and no-auth users) get full status
-            self.socketio.emit("status_update", status_data, to="admins")
+            socketio.emit("status_update", status_data, to="admins")
 
             # Each user room gets filtered status
             with self._rooms_lock:
@@ -147,32 +153,37 @@ def broadcast_status_update(self, status_data: dict[str, Any]) -> None:
 
     def _broadcast_status_update_to_room(self, room: str) -> None:
         """Broadcast status update to one user room."""
+        socketio = self._get_socketio()
+        if socketio is None:
+            return
+
         try:
             # Extract user_id from room name "user_123"
             uid = int(room.split("_", 1)[1])
             filtered = self._queue_status_fn(user_id=uid) if self._queue_status_fn else None
             if filtered is not None:
-                self.socketio.emit("status_update", filtered, to=room)
+                socketio.emit("status_update", filtered, to=room)
         except Exception:
             logger.exception("Failed to send status update for room %s", room)
 
     def broadcast_download_progress(
         self, book_id: str, progress: float, status: str, user_id: int | None = None
     ) -> None:
         """Broadcast download progress update for a specific book."""
-        if not self.is_enabled():
+        socketio = self._get_socketio()
+        if socketio is None:
             return
 
         try:
             data = {"book_id": book_id, "progress": progress, "status": status}
             # Admins always see all progress
-            self.socketio.emit("download_progress", data, to="admins")
+            socketio.emit("download_progress", data, to="admins")
             # If task belongs to a specific user, send to their room too
             if user_id is not None:
                 room = f"user_{user_id}"
                 with self._rooms_lock:
                     if room in self._user_rooms:
-                        self.socketio.emit("download_progress", data, to=room)
+                        socketio.emit("download_progress", data, to=room)
             logger.debug("Broadcasted progress for book %s: %s%%", book_id, progress)
         except Exception:
             logger.exception("Error broadcasting download progress")
@@ -186,7 +197,8 @@ def broadcast_search_status(
         phase: str = "searching",
     ) -> None:
         """Broadcast search status update for a release source search."""
-        if not self.is_enabled():
+        socketio = self._get_socketio()
+        if socketio is None:
             return
 
         try:
@@ -197,7 +209,7 @@ def broadcast_search_status(
                 "message": message,
                 "phase": phase,
             }
-            self.socketio.emit("search_status", data)
+            socketio.emit("search_status", data)
         except Exception:
             logger.exception("Error broadcasting search status")
 

shelfmark/bypass/external_bypasser.py

@@ -31,11 +31,30 @@
 BACKOFF_CAP = 10.0
 
 
+def _coerce_config_str(value: object, default: str) -> str:
+    """Return a string config value or a safe default."""
+    if isinstance(value, str):
+        return value
+    return default
+
+
+def _coerce_timeout_ms(value: object, default: int) -> int:
+    """Return a positive timeout in milliseconds or the default."""
+    if isinstance(value, bool):
+        return default
+    if isinstance(value, int) and value > 0:
+        return value
+    return default
+
+
 def _fetch_via_bypasser(target_url: str) -> str | None:
     """Make a single request to the external bypasser service. Returns HTML or None."""
-    raw_bypasser_url = config.get("EXT_BYPASSER_URL", "http://flaresolverr:8191")
-    bypasser_path = config.get("EXT_BYPASSER_PATH", "/v1")
-    bypasser_timeout = config.get("EXT_BYPASSER_TIMEOUT", 60000)
+    raw_bypasser_url = _coerce_config_str(
+        config.get("EXT_BYPASSER_URL", "http://flaresolverr:8191"),
+        "http://flaresolverr:8191",
+    )
+    bypasser_path = _coerce_config_str(config.get("EXT_BYPASSER_PATH", "/v1"), "/v1")
+    bypasser_timeout = _coerce_timeout_ms(config.get("EXT_BYPASSER_TIMEOUT", 60000), 60000)
 
     bypasser_url = normalize_http_url(raw_bypasser_url)
     if not bypasser_url or not bypasser_path:

shelfmark/bypass/internal_bypasser.py

@@ -17,7 +17,7 @@
 from http import HTTPStatus
 from pathlib import Path
 from threading import Event
-from typing import Any
+from typing import Any, Protocol, TypedDict, TypeGuard
 from urllib.parse import urlparse
 
 import requests
@@ -59,7 +59,21 @@
     "could not verify your browser automatically",
 ]
 
-DISPLAY = {
+
+class _DisplayState(TypedDict):
+    ffmpeg: subprocess.Popen[bytes] | None
+    ffmpeg_output: Path | None
+
+
+class _PageWithWindowRect(Protocol):
+    async def set_window_rect(self, x: int, _y: int, width: int, height: int) -> object: ...
+
+
+class _BrowserWithWindowRectPage(Protocol):
+    page: _PageWithWindowRect
+
+
+DISPLAY: _DisplayState = {
     "ffmpeg": None,
     "ffmpeg_output": None,
 }
@@ -95,6 +109,30 @@
 )
 
 
+def _coerce_positive_int(value: object, default: int) -> int:
+    """Return a positive integer config value or the provided default."""
+    if isinstance(value, bool):
+        return default
+    if isinstance(value, int) and value > 0:
+        return value
+    return default
+
+
+def _coerce_non_negative_float(value: object, default: float) -> float:
+    """Return a non-negative float config value or the provided default."""
+    if isinstance(value, bool):
+        return default
+    if isinstance(value, int | float) and value >= 0:
+        return float(value)
+    return default
+
+
+def _has_window_rect_page(candidate: object) -> TypeGuard[_BrowserWithWindowRectPage]:
+    """Check whether a browser wrapper exposes page.set_window_rect()."""
+    page = getattr(candidate, "page", None)
+    return callable(getattr(page, "set_window_rect", None))
+
+
 def _describe_runtime_path(path: str | Path) -> str:
     """Return compact ownership/mode info for a runtime path."""
     try:
@@ -613,7 +651,9 @@ async def _bypass(
     page: Any, max_retries: int | None = None, cancel_flag: Event | None = None
 ) -> bool:
     """Attempt to bypass Cloudflare/DDOS-Guard protection using multiple methods."""
-    max_retries = max_retries if max_retries is not None else app_config.MAX_RETRY
+    max_retries = (
+        max_retries if max_retries is not None else _coerce_positive_int(app_config.MAX_RETRY, 10)
+    )
 
     last_challenge_type = None
     consecutive_same_challenge = 0
@@ -790,7 +830,7 @@ async def _get(url: str, driver: Any, cancel_flag: Event | None = None) -> str:
 
 def get(url: str, retry: int | None = None, cancel_flag: Event | None = None) -> str:
     """Fetch a URL with protection bypass. Creates fresh Chrome instance for each bypass."""
-    retry = retry if retry is not None else app_config.MAX_RETRY
+    retry = retry if retry is not None else _coerce_positive_int(app_config.MAX_RETRY, 10)
 
     with LOCKED:
         # Try cookies first - another request may have completed bypass while waiting
@@ -879,16 +919,17 @@ async def _create_cdp_browser(url: str) -> Any:
         )
         raise
 
-    try:
-        await driver.page.set_window_rect(0, 0, screen_width, screen_height)
-    except _CDP_OPERATION_ERRORS as e:
-        logger.debug("Failed to set window size: %s", e)
+    if _has_window_rect_page(driver):
+        try:
+            await driver.page.set_window_rect(0, 0, screen_width, screen_height)
+        except _CDP_OPERATION_ERRORS as e:
+            logger.debug("Failed to set window size: %s", e)
 
     # Start FFmpeg recording if debug mode (record each bypass session)
     if app_config.get("DEBUG", False) and not DISPLAY.get("ffmpeg"):
         _start_ffmpeg_recording(display=os.environ.get("DISPLAY", ":0"))
 
-    await asyncio.sleep(app_config.DEFAULT_SLEEP)
+    await asyncio.sleep(_coerce_non_negative_float(app_config.DEFAULT_SLEEP, 5.0))
     logger.info("Chrome browser ready (Pure CDP)")
     logger.log_resource_usage()
     return driver

shelfmark/config/booklore_settings.py

@@ -123,7 +123,7 @@ def get_booklore_library_options() -> list[dict[str, Any]]:
 
     base_url = str(config.get("BOOKLORE_HOST", "") or "").strip().rstrip("/")
     username = str(config.get("BOOKLORE_USERNAME", "") or "").strip()
-    password = config.get("BOOKLORE_PASSWORD", "") or ""
+    password = str(config.get("BOOKLORE_PASSWORD", "") or "")
 
     if not base_url or not username or not password:
         return []
@@ -148,7 +148,7 @@ def get_booklore_path_options() -> list[dict[str, Any]]:
 
     base_url = str(config.get("BOOKLORE_HOST", "") or "").strip().rstrip("/")
     username = str(config.get("BOOKLORE_USERNAME", "") or "").strip()
-    password = config.get("BOOKLORE_PASSWORD", "") or ""
+    password = str(config.get("BOOKLORE_PASSWORD", "") or "")
 
     if not base_url or not username or not password:
         return []
@@ -182,7 +182,7 @@ def _get_value(key: str, default: object = None) -> object:
 
     base_url = str(_get_value("BOOKLORE_HOST", "") or "").strip().rstrip("/")
     username = str(_get_value("BOOKLORE_USERNAME", "") or "").strip()
-    password = _get_value("BOOKLORE_PASSWORD", "") or ""
+    password = str(_get_value("BOOKLORE_PASSWORD", "") or "")
 
     if not base_url:
         return {"success": False, "message": "Grimmory URL is required"}

shelfmark/config/env.py

@@ -79,7 +79,9 @@ def is_covers_cache_enabled() -> bool:
     from shelfmark.core.config import config
 
     setting_enabled = config.get("COVERS_CACHE_ENABLED", True)
-    return setting_enabled and _is_config_dir_writable()
+    if isinstance(setting_enabled, str):
+        return string_to_bool(setting_enabled) and _is_config_dir_writable()
+    return bool(setting_enabled) and _is_config_dir_writable()
 
 
 # =============================================================================

shelfmark/config/migrations.py

@@ -1,11 +1,14 @@
 """Configuration migration helpers."""
 
+from __future__ import annotations
+
 import json
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Protocol
 
 if TYPE_CHECKING:
     from collections.abc import Callable
+    from os import PathLike
 
 _DEPRECATED_SETTINGS_RESTRICTION_KEYS = (
     "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN",
@@ -14,6 +17,16 @@
 )
 
 
+class MigrationLogger(Protocol):
+    """Logger surface used by config migration helpers."""
+
+    def info(self, msg: str, *args: object) -> object: ...
+
+    def debug(self, msg: str, *args: object) -> object: ...
+
+    def exception(self, msg: str, *args: object) -> object: ...
+
+
 def _as_bool(value: object) -> bool:
     if isinstance(value, bool):
         return value
@@ -50,9 +63,9 @@ def migrate_security_settings(
     load_users_config: Callable[[], dict[str, Any]],
     save_users_config: Callable[[dict[str, Any]], None],
     ensure_config_dir: Callable[[], None],
-    get_config_path: Callable[[], object],
+    get_config_path: Callable[[], str | PathLike[str]],
     sync_builtin_admin_user: Callable[[str, str], None],
-    logger: object,
+    logger: MigrationLogger,
 ) -> None:
     """Migrate legacy security keys and sync builtin admin credentials."""
     try:

shelfmark/config/security.py

@@ -45,10 +45,13 @@ def _migrate_security_settings() -> None:
         save_config_file,
     )
 
+    def _save_users_config(values: dict[str, Any]) -> None:
+        save_config_file("users", values)
+
     migrate_security_settings(
         load_security_config=lambda: load_config_file("security"),
         load_users_config=lambda: load_config_file("users"),
-        save_users_config=lambda values: save_config_file("users", values),
+        save_users_config=_save_users_config,
         ensure_config_dir=lambda: _ensure_config_dir("security"),
         get_config_path=lambda: _get_config_file_path("security"),
         sync_builtin_admin_user=sync_builtin_admin_user,

shelfmark/config/settings.py

@@ -255,6 +255,11 @@ def _get_audiobook_release_source_options() -> list[dict[str, str]]:
 ]
 
 
+def _string_setting(value: object) -> str:
+    """Normalize free-form string settings used by select option builders."""
+    return value if isinstance(value, str) else str(value or "")
+
+
 def _get_aa_base_url_options() -> list[dict[str, str]]:
     """Build AA URL options dynamically, including additional mirrors from config."""
     from shelfmark.core.config import config
@@ -269,7 +274,7 @@ def _get_aa_base_url_options() -> list[dict[str, str]]:
     # If AA_BASE_URL is configured to a custom mirror that isn't present in the
     # defaults/additional list, include it so the UI can display the active value.
     configured_url = normalize_http_url(
-        config.get("AA_BASE_URL", "auto"),
+        _string_setting(config.get("AA_BASE_URL", "auto")),
         default_scheme="https",
         allow_special=("auto",),
     )
@@ -300,7 +305,7 @@ def _get_zlib_mirror_options() -> list[dict[str, str]]:
         options.append({"value": url, "label": domain})
 
     # Add custom mirrors
-    additional = config.get("ZLIB_ADDITIONAL_URLS", "")
+    additional = _string_setting(config.get("ZLIB_ADDITIONAL_URLS", ""))
     if additional:
         for raw_url in additional.split(","):
             url = raw_url.strip()
@@ -324,7 +329,7 @@ def _get_welib_mirror_options() -> list[dict[str, str]]:
         options.append({"value": url, "label": domain})
 
     # Add custom mirrors
-    additional = config.get("WELIB_ADDITIONAL_URLS", "")
+    additional = _string_setting(config.get("WELIB_ADDITIONAL_URLS", ""))
     if additional:
         for raw_url in additional.split(","):
             url = raw_url.strip()