refactor: move the secret key, session cookie, security headers and log filter into core/app_setup.py (#1639)

`main.py` also builds the app itself, and this moves the parts of that
which need no route into `core/app_setup.py`: the persisted secret key
and the session cookie settings, the security headers, and the filter
that keeps status polling and the WebSocket upgrade noise out of the
log. It is the app setup item in the plan from Discussion #1536.

The first commit adds 14 tests for them, since none existed: the key
reused, created with mode 600, replaced when too short, and still
returned when the config directory cannot be written, the cookie
settings, the headers on a real response and that a header a route set
itself is kept, and each case the log filter drops or keeps. They pass
on current `main`. The second commit is the move.

The security headers hook is registered from the same spot in `main.py`
as before, so the `after_request` order is unchanged. The secret key
function takes the config directory as a parameter instead of reading
`CONFIG_DIR`, which is the one change in the move, and the key tests
call it with a temp directory.

The URL map and the hook order are identical, and the full suite, ruff
and basedpyright are clean. I also ran the app from the branch: the
secret file is created with mode 600 and the four security headers come
back on a normal response.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
<!-- coderabbit-ai-generated:v1 -->

## Summary by CodeRabbit

* **Bug Fixes**
* Sessions now remain consistent across app restarts, and session
cookies use more protective browser settings, including HTTP-only access
and a seven-day lifetime.
* Responses include baseline security headers while preserving any
headers already set by the app.
* Routine status-polling and connection-upgrade messages are filtered
from application logs, reducing unrelated log noise.
* If saving a session key fails, the app can still start with a
generated key.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

shelfmark/core/app_setup.py

@@ -0,0 +1,138 @@
+"""App setup that does not depend on any route: the log filter, the persisted secret key with
+the session cookie settings, and the baseline security headers.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import TYPE_CHECKING
+
+from shelfmark.core.logger import setup_logger
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from flask import Flask
+    from werkzeug.wrappers import Response
+
+logger = setup_logger(__name__)
+
+FLASK_SECRET_KEY_MIN_BYTES = 32
+
+
+class LogNoiseFilter(logging.Filter):
+    """Filter out routine status endpoint requests and WebSocket upgrade errors to reduce log noise.
+
+    WebSocket upgrade errors are benign - Flask-SocketIO automatically falls back to polling transport.
+    The error occurs because Werkzeug's built-in server doesn't fully support WebSocket upgrades.
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Return whether a log record should be emitted."""
+        message = record.getMessage() if hasattr(record, "getMessage") else str(record.msg)
+
+        # Exclude GET /api/status requests (polling noise)
+        if "GET /api/status" in message:
+            return False
+
+        # Exclude WebSocket upgrade errors (benign - falls back to polling)
+        if "write() before start_response" in message:
+            return False
+
+        # Exclude the Error on request line that precedes WebSocket errors
+        if record.levelno == logging.ERROR:
+            if "Error on request:" in message:
+                return False
+            # Filter WebSocket-related AssertionError tracebacks
+            if hasattr(record, "exc_info") and record.exc_info:
+                exc_type, exc_value = record.exc_info[0], record.exc_info[1]
+                if (
+                    exc_type
+                    and exc_type.__name__ == "AssertionError"
+                    and exc_value
+                    and "write() before start_response" in str(exc_value)
+                ):
+                    return False
+
+        return True
+
+
+def configure_app_logging(app: Flask, app_logger: logging.Logger) -> None:
+    """Send Flask's and Werkzeug's logs through the app's handlers, minus the routine noise."""
+    app.logger.handlers = app_logger.handlers
+    app.logger.setLevel(app_logger.level)
+    # Also handle Werkzeug's logger
+    werkzeug_logger = logging.getLogger("werkzeug")
+    werkzeug_logger.handlers = app_logger.handlers
+    werkzeug_logger.setLevel(app_logger.level)
+    # Add filter to suppress routine status endpoint polling logs and WebSocket upgrade errors
+    werkzeug_logger.addFilter(LogNoiseFilter())
+
+
+def load_or_create_secret_key(config_dir: Path) -> bytes:
+    """Load a persisted Flask secret key from config, or create one."""
+    secret_path = config_dir / ".flask_secret"
+
+    try:
+        if secret_path.exists():
+            secret_key = secret_path.read_bytes()
+            if len(secret_key) >= FLASK_SECRET_KEY_MIN_BYTES:
+                return secret_key
+            logger.warning(
+                "Invalid persisted Flask secret key at %s (length=%s). Regenerating.",
+                secret_path,
+                len(secret_key),
+            )
+    except OSError as exc:
+        logger.warning("Failed to read Flask secret key at %s: %s", secret_path, exc)
+
+    secret_key = os.urandom(64)
+    try:
+        secret_path.parent.mkdir(parents=True, exist_ok=True)
+        secret_path.write_bytes(secret_key)
+        secret_path.chmod(0o600)
+    except OSError as exc:
+        logger.warning(
+            "Failed to persist Flask secret key at %s. Sessions may reset on restart: %s",
+            secret_path,
+            exc,
+        )
+
+    return secret_key
+
+
+def configure_secret_key_and_session(
+    app: Flask, config_dir: Path, *, cookie_secure: bool, cookie_name: str
+) -> None:
+    """Set the secret key and the session cookie settings."""
+    app.config.update(
+        SECRET_KEY=load_or_create_secret_key(config_dir),
+        SESSION_COOKIE_HTTPONLY=True,
+        SESSION_COOKIE_SAMESITE="Lax",
+        SESSION_COOKIE_SECURE=cookie_secure,
+        SESSION_COOKIE_NAME=cookie_name,
+        PERMANENT_SESSION_LIFETIME=604800,  # 7 days in seconds
+    )
+
+
+def register_security_headers(app: Flask) -> None:
+    """Add the baseline security headers to every response.
+
+    Call this at the place the hook should take in the ``after_request`` order, which runs
+    the hooks in reverse.
+    """
+
+    @app.after_request
+    def set_security_headers(response: Response) -> Response:
+        """Add baseline security headers to every response."""
+        response.headers.setdefault("X-Content-Type-Options", "nosniff")
+        response.headers.setdefault(
+            "Content-Security-Policy",
+            "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data: blob:; connect-src 'self' ws: wss:",
+        )
+        response.headers.setdefault(
+            "Permissions-Policy", "camera=(), microphone=(), geolocation=()"
+        )
+        response.headers.setdefault("Cross-Origin-Embedder-Policy", "credentialless")
+        return response

shelfmark/main.py

@@ -1,7 +1,6 @@
 """Flask app - routes, WebSocket handlers, and middleware."""
 
 import binascii
-import logging
 import os
 import re
 import sqlite3
@@ -42,6 +41,11 @@ from shelfmark.config.settings import (
 from shelfmark.core import api_key as api_key_module  # module access lets tests monkeypatch the key
 from shelfmark.core import search_deadline
 from shelfmark.core.activity_view_state_service import ActivityViewStateService
+from shelfmark.core.app_setup import (
+    configure_app_logging,
+    configure_secret_key_and_session,
+    register_security_headers,
+)
 from shelfmark.core.auth_modes import (
     get_auth_check_admin_status,
     is_settings_or_onboarding_path,
@@ -130,7 +134,6 @@ if TYPE_CHECKING:
     from shelfmark.metadata_providers import BookMetadata
 
 logger = setup_logger(__name__)
-FLASK_SECRET_KEY_MIN_BYTES = 32
 _OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error)
 _IMPORT_OPERATIONAL_ERRORS = (ImportError, *_OPERATIONAL_ERRORS)
 
@@ -633,97 +636,17 @@ if _is_debug_enabled():
     )
 
 
-# Custom log filter to exclude routine status endpoint polling and WebSocket noise
-class LogNoiseFilter(logging.Filter):
-    """Filter out routine status endpoint requests and WebSocket upgrade errors to reduce log noise.
-
-    WebSocket upgrade errors are benign - Flask-SocketIO automatically falls back to polling transport.
-    The error occurs because Werkzeug's built-in server doesn't fully support WebSocket upgrades.
-    """
-
-    def filter(self, record: logging.LogRecord) -> bool:
-        """Return whether a log record should be emitted."""
-        message = record.getMessage() if hasattr(record, "getMessage") else str(record.msg)
-
-        # Exclude GET /api/status requests (polling noise)
-        if "GET /api/status" in message:
-            return False
-
-        # Exclude WebSocket upgrade errors (benign - falls back to polling)
-        if "write() before start_response" in message:
-            return False
-
-        # Exclude the Error on request line that precedes WebSocket errors
-        if record.levelno == logging.ERROR:
-            if "Error on request:" in message:
-                return False
-            # Filter WebSocket-related AssertionError tracebacks
-            if hasattr(record, "exc_info") and record.exc_info:
-                exc_type, exc_value = record.exc_info[0], record.exc_info[1]
-                if (
-                    exc_type
-                    and exc_type.__name__ == "AssertionError"
-                    and exc_value
-                    and "write() before start_response" in str(exc_value)
-                ):
-                    return False
-
-        return True
-
-
-# Flask logger
-app.logger.handlers = logger.handlers
-app.logger.setLevel(logger.level)
-# Also handle Werkzeug's logger
-werkzeug_logger = logging.getLogger("werkzeug")
-werkzeug_logger.handlers = logger.handlers
-werkzeug_logger.setLevel(logger.level)
-# Add filter to suppress routine status endpoint polling logs and WebSocket upgrade errors
-werkzeug_logger.addFilter(LogNoiseFilter())
+configure_app_logging(app, logger)
 
 # Set up authentication defaults
 SESSION_COOKIE_SECURE = string_to_bool(SESSION_COOKIE_SECURE_ENV)
 
 
-def _load_or_create_secret_key() -> bytes:
-    """Load a persisted Flask secret key from config, or create one."""
-    secret_path = CONFIG_DIR / ".flask_secret"
-
-    try:
-        if secret_path.exists():
-            secret_key = secret_path.read_bytes()
-            if len(secret_key) >= FLASK_SECRET_KEY_MIN_BYTES:
-                return secret_key
-            logger.warning(
-                "Invalid persisted Flask secret key at %s (length=%s). Regenerating.",
-                secret_path,
-                len(secret_key),
-            )
-    except OSError as exc:
-        logger.warning("Failed to read Flask secret key at %s: %s", secret_path, exc)
-
-    secret_key = os.urandom(64)
-    try:
-        secret_path.parent.mkdir(parents=True, exist_ok=True)
-        secret_path.write_bytes(secret_key)
-        secret_path.chmod(0o600)
-    except OSError as exc:
-        logger.warning(
-            "Failed to persist Flask secret key at %s. Sessions may reset on restart: %s",
-            secret_path,
-            exc,
-        )
-
-    return secret_key
-
-
-app.config.update(
-    SECRET_KEY=_load_or_create_secret_key(),
-    SESSION_COOKIE_HTTPONLY=True,
-    SESSION_COOKIE_SAMESITE="Lax",
-    SESSION_COOKIE_SECURE=SESSION_COOKIE_SECURE,
-    SESSION_COOKIE_NAME=SESSION_COOKIE_NAME,
-    PERMANENT_SESSION_LIFETIME=604800,  # 7 days in seconds
+configure_secret_key_and_session(
+    app,
+    CONFIG_DIR,
+    cookie_secure=SESSION_COOKIE_SECURE,
+    cookie_name=SESSION_COOKIE_NAME,
 )
 configure_public_id_key(app.config["SECRET_KEY"])
 
@@ -949,17 +872,7 @@ def proxy_auth_middleware() -> Response | tuple[Response, int] | None:
         return None
 
 
-@app.after_request
-def set_security_headers(response: Response) -> Response:
-    """Add baseline security headers to every response."""
-    response.headers.setdefault("X-Content-Type-Options", "nosniff")
-    response.headers.setdefault(
-        "Content-Security-Policy",
-        "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data: blob:; connect-src 'self' ws: wss:",
-    )
-    response.headers.setdefault("Permissions-Policy", "camera=(), microphone=(), geolocation=()")
-    response.headers.setdefault("Cross-Origin-Embedder-Policy", "credentialless")
-    return response
+register_security_headers(app)
 
 
 @app.after_request

tests/core/test_app_setup.py

@@ -0,0 +1,155 @@
+"""Characterization tests for the app setup that moves out of main.py: the persisted secret key,
+the session cookie settings, the security headers and the werkzeug log filter.
+"""
+
+from __future__ import annotations
+
+import importlib
+import logging
+from unittest.mock import patch
+
+import pytest
+
+from shelfmark.core.app_setup import (
+    FLASK_SECRET_KEY_MIN_BYTES,
+    LogNoiseFilter,
+    load_or_create_secret_key,
+    register_security_headers,
+)
+
+
+@pytest.fixture(scope="module")
+def main_module():
+    """Import `shelfmark.main` with background startup disabled."""
+    with patch("shelfmark.download.orchestrator.start"):
+        import shelfmark.main as main
+
+        importlib.reload(main)
+        return main
+
+
+def _record(message: str, *, level: int = logging.INFO, exc_info=None) -> logging.LogRecord:
+    return logging.LogRecord("werkzeug", level, __file__, 1, message, None, exc_info)
+
+
+class TestSecretKey:
+    def test_a_persisted_key_is_reused(self, main_module, tmp_path):
+        persisted = b"k" * 64
+        (tmp_path / ".flask_secret").write_bytes(persisted)
+
+        assert load_or_create_secret_key(tmp_path) == persisted
+
+    def test_a_missing_key_is_created_and_kept_private(self, main_module, tmp_path):
+        created = load_or_create_secret_key(tmp_path)
+        again = load_or_create_secret_key(tmp_path)
+
+        path = tmp_path / ".flask_secret"
+        assert len(created) == 64
+        assert path.read_bytes() == created == again
+        assert path.stat().st_mode & 0o777 == 0o600
+
+    def test_a_key_that_is_too_short_is_replaced(self, main_module, tmp_path):
+        path = tmp_path / ".flask_secret"
+        path.write_bytes(b"short")
+
+        key = load_or_create_secret_key(tmp_path)
+
+        assert len(key) == 64
+        assert path.read_bytes() == key
+
+    def test_a_key_of_exactly_the_minimum_length_is_kept(self, main_module, tmp_path):
+        minimum = b"m" * FLASK_SECRET_KEY_MIN_BYTES
+        (tmp_path / ".flask_secret").write_bytes(minimum)
+
+        assert load_or_create_secret_key(tmp_path) == minimum
+
+    def test_an_unwritable_config_dir_still_gives_a_key(self, main_module, tmp_path):
+        blocked = tmp_path / "file-not-dir"
+        blocked.write_text("x")
+
+        key = load_or_create_secret_key(blocked / "config")
+
+        assert len(key) == 64
+
+    def test_the_app_and_the_opaque_id_key_use_the_same_secret(self, main_module):
+        from shelfmark.core import viewer_redaction
+
+        assert len(main_module.app.config["SECRET_KEY"]) >= 32
+        assert viewer_redaction._public_id_key is not None
+
+
+class TestSessionCookie:
+    def test_cookie_settings(self, main_module):
+        config = main_module.app.config
+
+        assert config["SESSION_COOKIE_HTTPONLY"] is True
+        assert config["SESSION_COOKIE_SAMESITE"] == "Lax"
+        assert config["SESSION_COOKIE_NAME"] == main_module.SESSION_COOKIE_NAME
+        assert config["SESSION_COOKIE_SECURE"] == main_module.SESSION_COOKIE_SECURE
+        assert config["PERMANENT_SESSION_LIFETIME"] == 604800
+
+
+class TestSecurityHeaders:
+    def test_every_response_carries_the_baseline_headers(self, main_module):
+        response = main_module.app.test_client().get("/api/health")
+
+        assert response.headers["X-Content-Type-Options"] == "nosniff"
+        assert response.headers["Permissions-Policy"] == "camera=(), microphone=(), geolocation=()"
+        assert response.headers["Cross-Origin-Embedder-Policy"] == "credentialless"
+        csp = response.headers["Content-Security-Policy"]
+        assert csp.startswith("default-src 'self'; script-src 'self';")
+        assert "connect-src 'self' ws: wss:" in csp
+
+    def test_a_header_already_on_the_response_is_not_overridden(self, main_module):
+        from flask import Flask, Response
+
+        response = Response("{}")
+        response.headers["X-Content-Type-Options"] = "route-value"
+
+        app = Flask(__name__)
+        register_security_headers(app)
+        with app.test_request_context():
+            result = app.process_response(response)
+
+        assert result is response
+        assert response.headers["X-Content-Type-Options"] == "route-value"
+        assert response.headers["Permissions-Policy"] == "camera=(), microphone=(), geolocation=()"
+
+
+class TestLogNoiseFilter:
+    def test_status_polling_is_dropped(self, main_module):
+        noise = LogNoiseFilter()
+
+        assert noise.filter(_record('127.0.0.1 - "GET /api/status HTTP/1.1" 200 -')) is False
+
+    def test_other_requests_are_kept(self, main_module):
+        noise = LogNoiseFilter()
+
+        assert noise.filter(_record('127.0.0.1 - "GET /api/releases HTTP/1.1" 200 -')) is True
+        assert noise.filter(_record('127.0.0.1 - "POST /api/status HTTP/1.1" 200 -')) is True
+
+    def test_the_websocket_upgrade_noise_is_dropped(self, main_module):
+        noise = LogNoiseFilter()
+
+        assert noise.filter(_record("write() before start_response")) is False
+        assert noise.filter(_record("Error on request:", level=logging.ERROR)) is False
+
+    def test_an_assertion_error_from_the_upgrade_is_dropped_but_other_errors_are_not(
+        self, main_module
+    ):
+        noise = LogNoiseFilter()
+        upgrade = AssertionError("write() before start_response")
+        other = AssertionError("something else")
+
+        dropped = _record("boom", level=logging.ERROR, exc_info=(AssertionError, upgrade, None))
+        kept = _record("boom", level=logging.ERROR, exc_info=(AssertionError, other, None))
+        kept_type = _record("boom", level=logging.ERROR, exc_info=(ValueError, upgrade, None))
+
+        assert noise.filter(dropped) is False
+        assert noise.filter(kept) is True
+        assert noise.filter(kept_type) is True
+
+    def test_the_filter_is_installed_on_the_werkzeug_logger(self, main_module):
+        filters = logging.getLogger("werkzeug").filters
+
+        assert any(type(f).__name__ == "LogNoiseFilter" for f in filters)