refactor: move the auth routes into core/auth_routes.py (#1625)

The auth routes are the next group from the plan in Discussion #1536:
`/api/auth/login`, `/logout` and `/check`, plus the login lockout
helpers, move from `main.py` to `core/auth_routes.py`, registered with
`register_auth_routes(app, ...)`. `main.py` is about 370 lines shorter.
This is the login path, so I kept it strictly to a move.

Two commits. The first adds 59 tests that pin the current behaviour
before anything moves: login in each mode (builtin, none, CWA and the
OIDC password fallback), wrong passwords, the lockout (the warning when
five attempts are left, the 429 and its text on the tenth failure,
per-user isolation, expiry), the local-auth switches, logout in each
mode and `/check`. They pass on current `main`, and I ran the same file
against `main` and against this branch. The second commit is the move,
and no existing test changed.

The lockout state is one dict, now living in `auth_routes`, and
`main.py` imports the same object so its other users (the
password-attempt limiter for the self-user routes, and the tests) see
it. Everything else that comes from `main` (the user database, the auth
mode, the client IP, the CWA database path, the local-auth settings and
the password check) is passed in as a callable evaluated per request, so
patches on `main` still apply.

Two things to know. `main.py` re-exports the lockout names and keeps
three lines assigning the view functions (`api_login =
app.view_functions["api_login"]` and the other two), because the e2e
tests call `main_module.api_login()` directly and I did not change tests
in this PR. If you would rather drop both, I can edit those e2e tests
instead. The other is that the lockout dict is no longer reset by a
reload of `main`, which only matters in tests, and they clear it
themselves.

The URL map is identical, all 77 rules, and the full suite, ruff and
basedpyright are clean. I also ran the app with builtin auth on a
scratch instance and went through it over HTTP: a wrong password gets a
401, login works and `/check` shows the session, logout clears it, the
tenth failed attempt gets the 429, and the right password is refused
while the account is locked.

shelfmark/core/auth_routes.py

@@ -0,0 +1,458 @@
+"""Sign-in routes: ``POST /api/auth/login``, ``POST /api/auth/logout``, ``GET /api/auth/check``.
+
+The failed-login counters and their lockout helpers live here too, as module state
+shared by everything that imports them (``main`` re-exports the names).
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING, Any
+
+from flask import jsonify, request, session
+
+from shelfmark.core.config import config as app_config
+from shelfmark.core.cwa_user_sync import upsert_cwa_user
+from shelfmark.core.logger import setup_logger
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+    from pathlib import Path
+
+    from flask import Flask
+    from werkzeug.wrappers import Response
+
+    from shelfmark.core.user_db import UserDB
+
+logger = setup_logger(__name__)
+
+# Rate limiting for login attempts
+# Map usernames to their failed-attempt counters and lockout timestamps.
+failed_login_attempts: dict[str, dict[str, Any]] = {}
+MAX_LOGIN_ATTEMPTS = 10
+LOCKOUT_DURATION_MINUTES = 30
+LOGIN_ATTEMPT_WARNING_THRESHOLD = 5
+
+
+def cleanup_old_lockouts() -> None:
+    """Remove expired lockout entries to prevent memory buildup."""
+    current_time = datetime.now(UTC)
+    expired_users = []
+    for username in list(failed_login_attempts):
+        lockout_until = _get_lockout_until(username, repair_if_locked=True)
+        if lockout_until is not None and lockout_until < current_time:
+            expired_users.append(username)
+    for username in expired_users:
+        logger.info("Lockout expired for user: %s", username)
+        del failed_login_attempts[username]
+
+
+def _get_lockout_until(username: str, *, repair_if_locked: bool = False) -> datetime | None:
+    """Return a valid lockout timestamp for the user when one exists.
+
+    When a user has already crossed the lockout threshold but the timestamp is
+    missing or malformed, optionally repair the state to keep the lockout in
+    force rather than silently letting the user through.
+    """
+    lockout_state = failed_login_attempts.get(username)
+    if lockout_state is None:
+        return None
+
+    lockout_until = lockout_state.get("lockout_until")
+    if isinstance(lockout_until, datetime):
+        return lockout_until
+
+    attempt_count = lockout_state.get("count")
+    if repair_if_locked and isinstance(attempt_count, int) and attempt_count >= MAX_LOGIN_ATTEMPTS:
+        repaired_lockout_until = datetime.now(UTC) + timedelta(minutes=LOCKOUT_DURATION_MINUTES)
+        lockout_state["lockout_until"] = repaired_lockout_until
+        logger.warning("Repaired missing lockout timestamp for locked account '%s'", username)
+        return repaired_lockout_until
+
+    if lockout_until is not None:
+        logger.warning("Ignoring invalid lockout timestamp for user '%s'", username)
+
+    return None
+
+
+def is_account_locked(username: str) -> bool:
+    """Check if an account is currently locked due to failed login attempts."""
+    cleanup_old_lockouts()
+
+    if username not in failed_login_attempts:
+        return False
+
+    lockout_until = _get_lockout_until(username, repair_if_locked=True)
+    return lockout_until is not None and datetime.now(UTC) < lockout_until
+
+
+def record_failed_login(username: str, ip_address: str) -> bool:
+    """Record a failed login attempt and lock account if threshold is reached.
+
+    Returns True if account is now locked, False otherwise.
+    """
+    if username not in failed_login_attempts:
+        failed_login_attempts[username] = {"count": 0}
+
+    failed_login_attempts[username]["count"] += 1
+    count = failed_login_attempts[username]["count"]
+
+    logger.warning(
+        "Failed login attempt %s/%s for user '%s' from IP %s",
+        count,
+        MAX_LOGIN_ATTEMPTS,
+        username,
+        ip_address,
+    )
+
+    if count >= MAX_LOGIN_ATTEMPTS:
+        lockout_until = datetime.now(UTC) + timedelta(minutes=LOCKOUT_DURATION_MINUTES)
+        failed_login_attempts[username]["lockout_until"] = lockout_until
+        logger.warning(
+            "Account locked for user '%s' until %s due to %s failed login attempts",
+            username,
+            lockout_until.strftime("%Y-%m-%d %H:%M:%S"),
+            count,
+        )
+        return True
+
+    return False
+
+
+def clear_failed_logins(username: str) -> None:
+    """Clear failed login attempts for a user after successful login."""
+    if username in failed_login_attempts:
+        del failed_login_attempts[username]
+        logger.debug("Cleared failed login attempts for user: %s", username)
+
+
+def _failed_login_response(username: str, ip_address: str) -> tuple[Response, int]:
+    """Handle a failed login attempt by recording it and returning the appropriate response."""
+    is_now_locked = record_failed_login(username, ip_address)
+
+    if is_now_locked:
+        return jsonify(
+            {
+                "error": f"Account locked due to {MAX_LOGIN_ATTEMPTS} failed login attempts. Try again in {LOCKOUT_DURATION_MINUTES} minutes."
+            }
+        ), 429
+
+    attempts_remaining = MAX_LOGIN_ATTEMPTS - failed_login_attempts[username]["count"]
+    if attempts_remaining <= LOGIN_ATTEMPT_WARNING_THRESHOLD:
+        return jsonify(
+            {"error": f"Invalid username or password. {attempts_remaining} attempts remaining."}
+        ), 401
+
+    return jsonify({"error": "Invalid username or password."}), 401
+
+
+def register_auth_routes(
+    app: Flask,
+    *,
+    get_user_db: Callable[[], UserDB | None],
+    resolve_auth_mode: Callable[[], str],
+    get_client_ip: Callable[[], str],
+    account_is_locked: Callable[[str], bool],
+    disable_local_auth: Callable[[], bool],
+    hide_local_auth: Callable[[], bool],
+    oidc_auto_redirect: Callable[[], bool],
+    get_cwa_db_path: Callable[[], Path | None],
+    check_password: Callable[[str, str], bool],
+    get_auth_check_admin_status: Callable[..., bool],
+    operational_errors: tuple[type[BaseException], ...],
+) -> None:
+    """Register the three auth routes.
+
+    Everything that lives in ``main.py`` or is patched there by tests is passed as a callable
+    that is called per request, so a runtime patch of any of them still takes effect.
+    ``account_is_locked`` is one of them: ``main.is_account_locked`` is looked up on each login.
+    """
+
+    @app.route("/api/auth/login", methods=["POST"])
+    def api_login() -> Response | tuple[Response, int]:
+        """Login endpoint that validates credentials and creates a session.
+
+        Supports both built-in credentials and CWA database authentication.
+        Includes rate limiting: 10 failed attempts = 30 minute lockout.
+
+        Request Body:
+            username (str): Username
+            password (str): Password
+            remember_me (bool): Whether to extend session duration
+
+        Returns:
+            flask.Response: JSON with success status or error message.
+
+        """
+        try:
+            ip_address = get_client_ip()
+            data = request.get_json(silent=True)
+            if not data:
+                return jsonify({"error": "No data provided"}), 400
+
+            auth_mode = resolve_auth_mode()
+            if auth_mode == "proxy":
+                return jsonify({"error": "Proxy authentication is enabled"}), 401
+
+            if auth_mode in ("builtin", "oidc") and disable_local_auth():
+                return jsonify({"error": "Local authentication is disabled"}), 403
+
+            if auth_mode == "oidc" and hide_local_auth():
+                return jsonify({"error": "Local authentication is disabled"}), 403
+
+            username = data.get("username", "").strip()
+            password = data.get("password", "")
+            remember_me = data.get("remember_me", False)
+
+            if not username or not password:
+                return jsonify({"error": "Username and password are required"}), 400
+
+            # Check if account is locked due to failed login attempts
+            if account_is_locked(username):
+                lockout_until = _get_lockout_until(username, repair_if_locked=True)
+                if lockout_until is None:
+                    logger.error("Locked account '%s' is missing a lockout timestamp", username)
+                    return jsonify(
+                        {
+                            "error": f"Account temporarily locked due to multiple failed login attempts. Try again in {LOCKOUT_DURATION_MINUTES} minutes."
+                        }
+                    ), 429
+                remaining_time = (lockout_until - datetime.now(UTC)).total_seconds() / 60
+                logger.warning(
+                    "Login attempt blocked for locked account '%s' from IP %s", username, ip_address
+                )
+                return jsonify(
+                    {
+                        "error": f"Account temporarily locked due to multiple failed login attempts. Try again in {int(remaining_time)} minutes."
+                    }
+                ), 429
+
+            # If no authentication is configured, authentication always succeeds
+            if auth_mode == "none":
+                session["user_id"] = username
+                session.permanent = remember_me
+                clear_failed_logins(username)
+                logger.info(
+                    "Login successful for user '%s' from IP %s (no auth configured)",
+                    username,
+                    ip_address,
+                )
+                return jsonify({"success": True})
+
+            # Password authentication (builtin and OIDC modes)
+            # OIDC mode also allows password login as a fallback so admins don't get locked out
+            if auth_mode in ("builtin", "oidc"):
+                user_db = get_user_db()
+                if user_db is None:
+                    logger.error("User database not available for %s auth", auth_mode)
+                    return jsonify({"error": "Authentication service unavailable"}), 503
+                try:
+                    db_user = user_db.get_user(username=username)
+
+                    if not db_user:
+                        return _failed_login_response(username, ip_address)
+
+                    # Authenticate against DB user
+                    if db_user:
+                        if not db_user.get("password_hash") or not check_password(
+                            db_user["password_hash"], password
+                        ):
+                            return _failed_login_response(username, ip_address)
+
+                        is_admin = db_user["role"] == "admin"
+                        session["user_id"] = username
+                        session["db_user_id"] = db_user["id"]
+                        session["is_admin"] = is_admin
+                        session.permanent = remember_me
+                        clear_failed_logins(username)
+                        logger.info(
+                            "Login successful for user '%s' from IP %s (%s auth, is_admin=%s, remember_me=%s)",
+                            username,
+                            ip_address,
+                            auth_mode,
+                            is_admin,
+                            remember_me,
+                        )
+                        return jsonify({"success": True})
+
+                    return _failed_login_response(username, ip_address)
+
+                except operational_errors as e:
+                    logger.error_trace(f"Built-in auth error: {e}")
+                    return jsonify({"error": "Authentication system error"}), 500
+
+            # CWA database authentication mode
+            if auth_mode == "cwa":
+                # Verify database still exists (it was validated at startup)
+                cwa_db_path = get_cwa_db_path()
+                if not cwa_db_path or not cwa_db_path.exists():
+                    logger.error("CWA database at %s is no longer accessible", cwa_db_path)
+                    return jsonify({"error": "Database configuration error"}), 500
+
+                try:
+                    db_path = os.fspath(cwa_db_path)
+                    db_uri = f"file:{db_path}?mode=ro&immutable=1"
+                    conn = sqlite3.connect(db_uri, uri=True)
+                    cur = conn.cursor()
+                    cur.execute(
+                        "SELECT password, role, email FROM user WHERE name = ?", (username,)
+                    )
+                    row = cur.fetchone()
+                    conn.close()
+
+                    # Check if user exists and password is correct
+                    if not row or not row[0] or not check_password(row[0], password):
+                        return _failed_login_response(username, ip_address)
+
+                    # Check if user has admin role (ROLE_ADMIN = 1, bit flag)
+                    user_role = row[1] if row[1] is not None else 0
+                    is_admin = (user_role & 1) == 1
+                    cwa_email = row[2] or None
+
+                    db_user_id = None
+                    user_db = get_user_db()
+                    if user_db is not None:
+                        role = "admin" if is_admin else "user"
+                        db_user, _ = upsert_cwa_user(
+                            user_db,
+                            cwa_username=username,
+                            cwa_email=cwa_email,
+                            role=role,
+                            context="cwa_login",
+                        )
+                        db_user_id = db_user["id"]
+
+                    # Successful authentication - create session and clear failed attempts
+                    session["user_id"] = username
+                    session["is_admin"] = is_admin
+                    if db_user_id is not None:
+                        session["db_user_id"] = db_user_id
+                    session.permanent = remember_me
+                    clear_failed_logins(username)
+                    logger.info(
+                        "Login successful for user '%s' from IP %s (CWA auth, is_admin=%s, remember_me=%s)",
+                        username,
+                        ip_address,
+                        is_admin,
+                        remember_me,
+                    )
+                    return jsonify({"success": True})
+
+                except operational_errors as e:
+                    logger.error_trace(f"CWA database error during login: {e}")
+                    return jsonify({"error": "Authentication system error"}), 500
+
+            # Should not reach here, but handle gracefully
+            return jsonify({"error": "Unknown authentication mode"}), 500
+
+        except operational_errors as e:
+            logger.error_trace(f"Login error: {e}")
+            return jsonify({"error": "Login failed"}), 500
+
+    @app.route("/api/auth/logout", methods=["POST"])
+    def api_logout() -> Response | tuple[Response, int]:
+        """Logout endpoint that clears the session.
+
+        For proxy auth, returns the logout URL if configured.
+
+        Returns:
+            flask.Response: JSON with success status and optional logout_url.
+
+        """
+        try:
+            auth_mode = resolve_auth_mode()
+            ip_address = get_client_ip()
+            username = session.get("user_id", "unknown")
+            session.clear()
+            logger.info("Logout successful for user '%s' from IP %s", username, ip_address)
+
+            # For proxy auth, include logout URL if configured
+            if auth_mode == "proxy":
+                logout_url = app_config.get("PROXY_AUTH_LOGOUT_URL", "")
+                if logout_url:
+                    return jsonify({"success": True, "logout_url": logout_url})
+
+            return jsonify({"success": True})
+        except operational_errors as e:
+            logger.error_trace(f"Logout error: {e}")
+            return jsonify({"error": "Logout failed"}), 500
+
+    @app.route("/api/auth/check", methods=["GET"])
+    def api_auth_check() -> Response | tuple[Response, int]:
+        """Check if user has a valid session.
+
+        Returns:
+            flask.Response: JSON with authentication status, whether auth is required,
+            which auth mode is active, and whether user has admin privileges.
+
+        """
+        try:
+            auth_mode = resolve_auth_mode()
+
+            # If no authentication is configured, access is allowed (full admin)
+            if auth_mode == "none":
+                return jsonify(
+                    {
+                        "authenticated": True,
+                        "auth_required": False,
+                        "auth_mode": "none",
+                        "is_admin": True,
+                    }
+                )
+
+            # Check if user has a valid session
+            is_authenticated = "user_id" in session
+
+            is_admin = get_auth_check_admin_status(auth_mode, {}, session)
+
+            display_name = None
+            user_db = get_user_db()
+            if is_authenticated and session.get("db_user_id") and user_db is not None:
+                try:
+                    db_user = user_db.get_user(user_id=session["db_user_id"])
+                    if db_user:
+                        display_name = db_user.get("display_name") or None
+                except (sqlite3.Error, TypeError, ValueError) as exc:
+                    logger.debug("Could not load display name for session user: %s", exc)
+
+            response_data = {
+                "authenticated": is_authenticated,
+                "auth_required": True,
+                "auth_mode": auth_mode,
+                "is_admin": is_admin if is_authenticated else False,
+                "username": session.get("user_id") if is_authenticated else None,
+                "display_name": display_name,
+            }
+
+            # Add logout URL for proxy auth if configured
+            if auth_mode == "proxy" and app_config.get("PROXY_AUTH_USER_HEADER", ""):
+                logout_url = app_config.get("PROXY_AUTH_LOGOUT_URL", "")
+                if logout_url:
+                    response_data["logout_url"] = logout_url
+
+            if auth_mode in ("builtin", "oidc") and disable_local_auth():
+                response_data["hide_local_auth"] = True
+
+            # Add custom OIDC button label and SSO enforcement flags if configured
+            if auth_mode == "oidc":
+                oidc_button_label = app_config.get("OIDC_BUTTON_LABEL", "")
+                if oidc_button_label:
+                    response_data["oidc_button_label"] = oidc_button_label
+                if hide_local_auth():
+                    response_data["hide_local_auth"] = True
+                if oidc_auto_redirect():
+                    response_data["oidc_auto_redirect"] = True
+
+            return jsonify(response_data)
+        except operational_errors as e:
+            logger.error_trace(f"Auth check error: {e}")
+            return jsonify(
+                {
+                    "authenticated": False,
+                    "auth_required": True,
+                    "auth_mode": "unknown",
+                    "is_admin": False,
+                }
+            )

shelfmark/main.py

@@ -7,7 +7,6 @@ import re
 import sqlite3
 import time
 import unicodedata
-from datetime import UTC, datetime, timedelta
 from functools import wraps
 from importlib import import_module
 from pathlib import Path
@@ -49,8 +48,19 @@ from shelfmark.core.auth_modes import (
     load_active_auth_mode,
     requires_admin_for_settings_access,
 )
+
+# The lockout names are re-exported: tests and other callers reach them as `shelfmark.main.<name>`.
+from shelfmark.core.auth_routes import (  # noqa: F401
+    LOCKOUT_DURATION_MINUTES,
+    LOGIN_ATTEMPT_WARNING_THRESHOLD,
+    MAX_LOGIN_ATTEMPTS,
+    clear_failed_logins,
+    failed_login_attempts,
+    is_account_locked,
+    record_failed_login,
+    register_auth_routes,
+)
 from shelfmark.core.config import config as app_config
-from shelfmark.core.cwa_user_sync import upsert_cwa_user
 from shelfmark.core.download_history_service import DownloadHistoryService
 from shelfmark.core.external_user_linking import upsert_external_user
 from shelfmark.core.logger import setup_logger
@@ -305,105 +315,6 @@ if user_db is not None:
 # first user search does not pay for a cold Chrome bypass. Never blocks startup.
 warmup.start()
 
-# Rate limiting for login attempts
-# Map usernames to their failed-attempt counters and lockout timestamps.
-failed_login_attempts: dict[str, dict[str, Any]] = {}
-MAX_LOGIN_ATTEMPTS = 10
-LOCKOUT_DURATION_MINUTES = 30
-LOGIN_ATTEMPT_WARNING_THRESHOLD = 5
-
-
-def cleanup_old_lockouts() -> None:
-    """Remove expired lockout entries to prevent memory buildup."""
-    current_time = datetime.now(UTC)
-    expired_users = []
-    for username in list(failed_login_attempts):
-        lockout_until = _get_lockout_until(username, repair_if_locked=True)
-        if lockout_until is not None and lockout_until < current_time:
-            expired_users.append(username)
-    for username in expired_users:
-        logger.info("Lockout expired for user: %s", username)
-        del failed_login_attempts[username]
-
-
-def _get_lockout_until(username: str, *, repair_if_locked: bool = False) -> datetime | None:
-    """Return a valid lockout timestamp for the user when one exists.
-
-    When a user has already crossed the lockout threshold but the timestamp is
-    missing or malformed, optionally repair the state to keep the lockout in
-    force rather than silently letting the user through.
-    """
-    lockout_state = failed_login_attempts.get(username)
-    if lockout_state is None:
-        return None
-
-    lockout_until = lockout_state.get("lockout_until")
-    if isinstance(lockout_until, datetime):
-        return lockout_until
-
-    attempt_count = lockout_state.get("count")
-    if repair_if_locked and isinstance(attempt_count, int) and attempt_count >= MAX_LOGIN_ATTEMPTS:
-        repaired_lockout_until = datetime.now(UTC) + timedelta(minutes=LOCKOUT_DURATION_MINUTES)
-        lockout_state["lockout_until"] = repaired_lockout_until
-        logger.warning("Repaired missing lockout timestamp for locked account '%s'", username)
-        return repaired_lockout_until
-
-    if lockout_until is not None:
-        logger.warning("Ignoring invalid lockout timestamp for user '%s'", username)
-
-    return None
-
-
-def is_account_locked(username: str) -> bool:
-    """Check if an account is currently locked due to failed login attempts."""
-    cleanup_old_lockouts()
-
-    if username not in failed_login_attempts:
-        return False
-
-    lockout_until = _get_lockout_until(username, repair_if_locked=True)
-    return lockout_until is not None and datetime.now(UTC) < lockout_until
-
-
-def record_failed_login(username: str, ip_address: str) -> bool:
-    """Record a failed login attempt and lock account if threshold is reached.
-
-    Returns True if account is now locked, False otherwise.
-    """
-    if username not in failed_login_attempts:
-        failed_login_attempts[username] = {"count": 0}
-
-    failed_login_attempts[username]["count"] += 1
-    count = failed_login_attempts[username]["count"]
-
-    logger.warning(
-        "Failed login attempt %s/%s for user '%s' from IP %s",
-        count,
-        MAX_LOGIN_ATTEMPTS,
-        username,
-        ip_address,
-    )
-
-    if count >= MAX_LOGIN_ATTEMPTS:
-        lockout_until = datetime.now(UTC) + timedelta(minutes=LOCKOUT_DURATION_MINUTES)
-        failed_login_attempts[username]["lockout_until"] = lockout_until
-        logger.warning(
-            "Account locked for user '%s' until %s due to %s failed login attempts",
-            username,
-            lockout_until.strftime("%Y-%m-%d %H:%M:%S"),
-            count,
-        )
-        return True
-
-    return False
-
-
-def clear_failed_logins(username: str) -> None:
-    """Clear failed login attempts for a user after successful login."""
-    if username in failed_login_attempts:
-        del failed_login_attempts[username]
-        logger.debug("Cleared failed login attempts for user: %s", username)
-
 
 def get_client_ip() -> str:
     """Extract client IP address from request, handling reverse proxy forwarding."""
@@ -2184,308 +2095,26 @@ def internal_error(error: Exception) -> Response | tuple[Response, int]:
     return jsonify({"error": "Internal server error"}), 500
 
 
-def _failed_login_response(username: str, ip_address: str) -> tuple[Response, int]:
-    """Handle a failed login attempt by recording it and returning the appropriate response."""
-    is_now_locked = record_failed_login(username, ip_address)
-
-    if is_now_locked:
-        return jsonify(
-            {
-                "error": f"Account locked due to {MAX_LOGIN_ATTEMPTS} failed login attempts. Try again in {LOCKOUT_DURATION_MINUTES} minutes."
-            }
-        ), 429
-
-    attempts_remaining = MAX_LOGIN_ATTEMPTS - failed_login_attempts[username]["count"]
-    if attempts_remaining <= LOGIN_ATTEMPT_WARNING_THRESHOLD:
-        return jsonify(
-            {"error": f"Invalid username or password. {attempts_remaining} attempts remaining."}
-        ), 401
-
-    return jsonify({"error": "Invalid username or password."}), 401
-
-
-@app.route("/api/auth/login", methods=["POST"])
-def api_login() -> Response | tuple[Response, int]:
-    """Login endpoint that validates credentials and creates a session.
-
-    Supports both built-in credentials and CWA database authentication.
-    Includes rate limiting: 10 failed attempts = 30 minute lockout.
-
-    Request Body:
-        username (str): Username
-        password (str): Password
-        remember_me (bool): Whether to extend session duration
-
-    Returns:
-        flask.Response: JSON with success status or error message.
-
-    """
-    try:
-        ip_address = get_client_ip()
-        data = request.get_json(silent=True)
-        if not data:
-            return jsonify({"error": "No data provided"}), 400
-
-        auth_mode = get_auth_mode()
-        if auth_mode == "proxy":
-            return jsonify({"error": "Proxy authentication is enabled"}), 401
-
-        if auth_mode in ("builtin", "oidc") and DISABLE_LOCAL_AUTH:
-            return jsonify({"error": "Local authentication is disabled"}), 403
-
-        if auth_mode == "oidc" and HIDE_LOCAL_AUTH:
-            return jsonify({"error": "Local authentication is disabled"}), 403
-
-        username = data.get("username", "").strip()
-        password = data.get("password", "")
-        remember_me = data.get("remember_me", False)
-
-        if not username or not password:
-            return jsonify({"error": "Username and password are required"}), 400
-
-        # Check if account is locked due to failed login attempts
-        if is_account_locked(username):
-            lockout_until = _get_lockout_until(username, repair_if_locked=True)
-            if lockout_until is None:
-                logger.error("Locked account '%s' is missing a lockout timestamp", username)
-                return jsonify(
-                    {
-                        "error": f"Account temporarily locked due to multiple failed login attempts. Try again in {LOCKOUT_DURATION_MINUTES} minutes."
-                    }
-                ), 429
-            remaining_time = (lockout_until - datetime.now(UTC)).total_seconds() / 60
-            logger.warning(
-                "Login attempt blocked for locked account '%s' from IP %s", username, ip_address
-            )
-            return jsonify(
-                {
-                    "error": f"Account temporarily locked due to multiple failed login attempts. Try again in {int(remaining_time)} minutes."
-                }
-            ), 429
-
-        # If no authentication is configured, authentication always succeeds
-        if auth_mode == "none":
-            session["user_id"] = username
-            session.permanent = remember_me
-            clear_failed_logins(username)
-            logger.info(
-                "Login successful for user '%s' from IP %s (no auth configured)",
-                username,
-                ip_address,
-            )
-            return jsonify({"success": True})
-
-        # Password authentication (builtin and OIDC modes)
-        # OIDC mode also allows password login as a fallback so admins don't get locked out
-        if auth_mode in ("builtin", "oidc"):
-            if user_db is None:
-                logger.error("User database not available for %s auth", auth_mode)
-                return jsonify({"error": "Authentication service unavailable"}), 503
-            try:
-                db_user = user_db.get_user(username=username)
-
-                if not db_user:
-                    return _failed_login_response(username, ip_address)
-
-                # Authenticate against DB user
-                if db_user:
-                    if not db_user.get("password_hash") or not check_password_hash(
-                        db_user["password_hash"], password
-                    ):
-                        return _failed_login_response(username, ip_address)
-
-                    is_admin = db_user["role"] == "admin"
-                    session["user_id"] = username
-                    session["db_user_id"] = db_user["id"]
-                    session["is_admin"] = is_admin
-                    session.permanent = remember_me
-                    clear_failed_logins(username)
-                    logger.info(
-                        "Login successful for user '%s' from IP %s (%s auth, is_admin=%s, remember_me=%s)",
-                        username,
-                        ip_address,
-                        auth_mode,
-                        is_admin,
-                        remember_me,
-                    )
-                    return jsonify({"success": True})
-
-                return _failed_login_response(username, ip_address)
-
-            except _OPERATIONAL_ERRORS as e:
-                logger.error_trace(f"Built-in auth error: {e}")
-                return jsonify({"error": "Authentication system error"}), 500
-
-        # CWA database authentication mode
-        if auth_mode == "cwa":
-            # Verify database still exists (it was validated at startup)
-            if not CWA_DB_PATH or not CWA_DB_PATH.exists():
-                logger.error("CWA database at %s is no longer accessible", CWA_DB_PATH)
-                return jsonify({"error": "Database configuration error"}), 500
-
-            try:
-                db_path = os.fspath(CWA_DB_PATH)
-                db_uri = f"file:{db_path}?mode=ro&immutable=1"
-                conn = sqlite3.connect(db_uri, uri=True)
-                cur = conn.cursor()
-                cur.execute("SELECT password, role, email FROM user WHERE name = ?", (username,))
-                row = cur.fetchone()
-                conn.close()
-
-                # Check if user exists and password is correct
-                if not row or not row[0] or not check_password_hash(row[0], password):
-                    return _failed_login_response(username, ip_address)
-
-                # Check if user has admin role (ROLE_ADMIN = 1, bit flag)
-                user_role = row[1] if row[1] is not None else 0
-                is_admin = (user_role & 1) == 1
-                cwa_email = row[2] or None
-
-                db_user_id = None
-                if user_db is not None:
-                    role = "admin" if is_admin else "user"
-                    db_user, _ = upsert_cwa_user(
-                        user_db,
-                        cwa_username=username,
-                        cwa_email=cwa_email,
-                        role=role,
-                        context="cwa_login",
-                    )
-                    db_user_id = db_user["id"]
-
-                # Successful authentication - create session and clear failed attempts
-                session["user_id"] = username
-                session["is_admin"] = is_admin
-                if db_user_id is not None:
-                    session["db_user_id"] = db_user_id
-                session.permanent = remember_me
-                clear_failed_logins(username)
-                logger.info(
-                    "Login successful for user '%s' from IP %s (CWA auth, is_admin=%s, remember_me=%s)",
-                    username,
-                    ip_address,
-                    is_admin,
-                    remember_me,
-                )
-                return jsonify({"success": True})
-
-            except _OPERATIONAL_ERRORS as e:
-                logger.error_trace(f"CWA database error during login: {e}")
-                return jsonify({"error": "Authentication system error"}), 500
-
-        # Should not reach here, but handle gracefully
-        return jsonify({"error": "Unknown authentication mode"}), 500
-
-    except _OPERATIONAL_ERRORS as e:
-        logger.error_trace(f"Login error: {e}")
-        return jsonify({"error": "Login failed"}), 500
-
-
-@app.route("/api/auth/logout", methods=["POST"])
-def api_logout() -> Response | tuple[Response, int]:
-    """Logout endpoint that clears the session.
-
-    For proxy auth, returns the logout URL if configured.
-
-    Returns:
-        flask.Response: JSON with success status and optional logout_url.
-
-    """
-    try:
-        auth_mode = get_auth_mode()
-        ip_address = get_client_ip()
-        username = session.get("user_id", "unknown")
-        session.clear()
-        logger.info("Logout successful for user '%s' from IP %s", username, ip_address)
-
-        # For proxy auth, include logout URL if configured
-        if auth_mode == "proxy":
-            logout_url = app_config.get("PROXY_AUTH_LOGOUT_URL", "")
-            if logout_url:
-                return jsonify({"success": True, "logout_url": logout_url})
-
-        return jsonify({"success": True})
-    except _OPERATIONAL_ERRORS as e:
-        logger.error_trace(f"Logout error: {e}")
-        return jsonify({"error": "Logout failed"}), 500
-
-
-@app.route("/api/auth/check", methods=["GET"])
-def api_auth_check() -> Response | tuple[Response, int]:
-    """Check if user has a valid session.
-
-    Returns:
-        flask.Response: JSON with authentication status, whether auth is required,
-        which auth mode is active, and whether user has admin privileges.
-
-    """
-    try:
-        auth_mode = get_auth_mode()
-
-        # If no authentication is configured, access is allowed (full admin)
-        if auth_mode == "none":
-            return jsonify(
-                {
-                    "authenticated": True,
-                    "auth_required": False,
-                    "auth_mode": "none",
-                    "is_admin": True,
-                }
-            )
-
-        # Check if user has a valid session
-        is_authenticated = "user_id" in session
-
-        is_admin = get_auth_check_admin_status(auth_mode, {}, session)
-
-        display_name = None
-        if is_authenticated and session.get("db_user_id") and user_db is not None:
-            try:
-                db_user = user_db.get_user(user_id=session["db_user_id"])
-                if db_user:
-                    display_name = db_user.get("display_name") or None
-            except (sqlite3.Error, TypeError, ValueError) as exc:
-                logger.debug("Could not load display name for session user: %s", exc)
-
-        response_data = {
-            "authenticated": is_authenticated,
-            "auth_required": True,
-            "auth_mode": auth_mode,
-            "is_admin": is_admin if is_authenticated else False,
-            "username": session.get("user_id") if is_authenticated else None,
-            "display_name": display_name,
-        }
-
-        # Add logout URL for proxy auth if configured
-        if auth_mode == "proxy" and app_config.get("PROXY_AUTH_USER_HEADER", ""):
-            logout_url = app_config.get("PROXY_AUTH_LOGOUT_URL", "")
-            if logout_url:
-                response_data["logout_url"] = logout_url
-
-        if auth_mode in ("builtin", "oidc") and DISABLE_LOCAL_AUTH:
-            response_data["hide_local_auth"] = True
-
-        # Add custom OIDC button label and SSO enforcement flags if configured
-        if auth_mode == "oidc":
-            oidc_button_label = app_config.get("OIDC_BUTTON_LABEL", "")
-            if oidc_button_label:
-                response_data["oidc_button_label"] = oidc_button_label
-            if HIDE_LOCAL_AUTH:
-                response_data["hide_local_auth"] = True
-            if OIDC_AUTO_REDIRECT:
-                response_data["oidc_auto_redirect"] = True
-
-        return jsonify(response_data)
-    except _OPERATIONAL_ERRORS as e:
-        logger.error_trace(f"Auth check error: {e}")
-        return jsonify(
-            {
-                "authenticated": False,
-                "auth_required": True,
-                "auth_mode": "unknown",
-                "is_admin": False,
-            }
-        )
+register_auth_routes(
+    app,
+    get_user_db=lambda: user_db,
+    resolve_auth_mode=_resolve_auth_mode_for_routes,
+    get_client_ip=lambda: get_client_ip(),  # noqa: PLW0108
+    account_is_locked=lambda username: is_account_locked(username),  # noqa: PLW0108
+    disable_local_auth=lambda: DISABLE_LOCAL_AUTH,
+    hide_local_auth=lambda: HIDE_LOCAL_AUTH,
+    oidc_auto_redirect=lambda: OIDC_AUTO_REDIRECT,
+    get_cwa_db_path=lambda: CWA_DB_PATH,
+    check_password=lambda pwhash, password: check_password_hash(pwhash, password),  # noqa: PLW0108
+    get_auth_check_admin_status=lambda auth_mode, headers, sess: get_auth_check_admin_status(  # noqa: PLW0108
+        auth_mode, headers, sess
+    ),
+    operational_errors=_OPERATIONAL_ERRORS,
+)
+# The views as plain module attributes, for tests/e2e/test_auth_endpoints.py, which calls them directly.
+api_login = app.view_functions["api_login"]
+api_logout = app.view_functions["api_logout"]
+api_auth_check = app.view_functions["api_auth_check"]
 
 
 def _attach_library_ownership(book: Any, book_dict: dict[str, Any]) -> None:

tests/core/test_auth_routes.py

@@ -0,0 +1,674 @@
+"""Characterization tests for `/api/auth/login`, `/api/auth/logout` and `/api/auth/check`.
+
+They pin the behaviour the three routes have today (lockout thresholds and timing,
+messages, status codes, session writes, one case per auth mode), so a refactor that
+moves the routes can be checked against them. Every credential here is made up.
+"""
+
+from __future__ import annotations
+
+import importlib
+import sqlite3
+from datetime import UTC, datetime, timedelta
+from unittest.mock import patch
+
+import pytest
+from werkzeug.security import generate_password_hash
+
+
+@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
+
+
+@pytest.fixture
+def client(main_module):
+    main_module.failed_login_attempts.clear()
+    try:
+        yield main_module.app.test_client()
+    finally:
+        main_module.failed_login_attempts.clear()
+
+
+@pytest.fixture
+def temp_user_db(tmp_path, main_module, monkeypatch):
+    from shelfmark.core.user_db import UserDB
+
+    db = UserDB(str(tmp_path / "users.db"))
+    db.initialize()
+    monkeypatch.setattr(main_module, "user_db", db)
+    return db
+
+
+def _mode(main_module, mode):
+    return patch.object(main_module, "get_auth_mode", return_value=mode)
+
+
+def _login(client, username="alice", password="fake-1", **extra):
+    return client.post(
+        "/api/auth/login", json={"username": username, "password": password, **extra}
+    )
+
+
+def _make_alice(db, role="user"):
+    return db.create_user(
+        username="alice", password_hash=generate_password_hash("fake-1"), role=role
+    )
+
+
+class TestLoginInput:
+    def test_empty_json_object_is_no_data(self, main_module, client):
+        with _mode(main_module, "builtin"):
+            response = client.post("/api/auth/login", json={})
+        assert response.status_code == 400
+        assert response.get_json() == {"error": "No data provided"}
+
+    @pytest.mark.parametrize(
+        "payload",
+        [
+            {"username": "", "password": "x"},
+            {"username": "   ", "password": "x"},
+            {"username": "alice", "password": ""},
+            {"username": "alice"},
+            {"password": "x"},
+        ],
+    )
+    def test_missing_username_or_password(self, main_module, client, payload):
+        with _mode(main_module, "builtin"):
+            response = client.post("/api/auth/login", json=payload)
+        assert response.status_code == 400
+        assert response.get_json() == {"error": "Username and password are required"}
+        assert main_module.failed_login_attempts == {}
+
+    def test_username_is_stripped(self, main_module, client, temp_user_db):
+        _make_alice(temp_user_db)
+        with _mode(main_module, "builtin"):
+            response = _login(client, username="  alice  ")
+        assert response.status_code == 200
+        with client.session_transaction() as sess:
+            assert sess["user_id"] == "alice"
+
+    def test_proxy_mode_wins_over_missing_data_checks_order(self, main_module, client):
+        # Proxy mode answers before the username/password check.
+        with _mode(main_module, "proxy"):
+            response = client.post("/api/auth/login", json={"username": "", "password": ""})
+        assert response.status_code == 401
+        assert response.get_json() == {"error": "Proxy authentication is enabled"}
+
+
+class TestLoginBuiltin:
+    def test_success_sets_session_cookie_and_clears_failures(
+        self, main_module, client, temp_user_db
+    ):
+        user = _make_alice(temp_user_db, role="user")
+        main_module.failed_login_attempts["alice"] = {"count": 3}
+        with _mode(main_module, "builtin"):
+            response = _login(client, remember_me=True)
+        assert response.status_code == 200
+        assert response.get_json() == {"success": True}
+        assert any(
+            header.startswith(main_module.app.config["SESSION_COOKIE_NAME"] + "=")
+            for header in response.headers.getlist("Set-Cookie")
+        )
+        with client.session_transaction() as sess:
+            assert sess["user_id"] == "alice"
+            assert sess["db_user_id"] == user["id"]
+            assert sess["is_admin"] is False
+            assert sess.permanent is True
+        assert "alice" not in main_module.failed_login_attempts
+
+    def test_oidc_mode_allows_password_fallback(self, main_module, client, temp_user_db):
+        _make_alice(temp_user_db, role="admin")
+        with _mode(main_module, "oidc"):
+            response = _login(client)
+        assert response.status_code == 200
+        with client.session_transaction() as sess:
+            assert sess["is_admin"] is True
+
+    def test_unknown_user_is_counted_and_gets_the_generic_message(
+        self, main_module, client, temp_user_db
+    ):
+        with _mode(main_module, "builtin"):
+            response = _login(client, username="nobody")
+        assert response.status_code == 401
+        assert response.get_json() == {"error": "Invalid username or password."}
+        assert main_module.failed_login_attempts["nobody"]["count"] == 1
+        with client.session_transaction() as sess:
+            assert "user_id" not in sess
+
+    def test_user_without_password_hash_is_rejected(self, main_module, client, temp_user_db):
+        temp_user_db.create_user(username="alice", role="user")
+        with _mode(main_module, "builtin"):
+            response = _login(client)
+        assert response.status_code == 401
+        assert main_module.failed_login_attempts["alice"]["count"] == 1
+
+    def test_wrong_password_writes_no_session(self, main_module, client, temp_user_db):
+        _make_alice(temp_user_db)
+        with _mode(main_module, "builtin"):
+            response = _login(client, password="wrong")
+        assert response.status_code == 401
+        with client.session_transaction() as sess:
+            assert "user_id" not in sess
+            assert "is_admin" not in sess
+
+    def test_no_user_database_is_503(self, main_module, client, monkeypatch):
+        monkeypatch.setattr(main_module, "user_db", None)
+        with _mode(main_module, "builtin"):
+            response = _login(client)
+        assert response.status_code == 503
+        assert response.get_json() == {"error": "Authentication service unavailable"}
+
+    def test_database_error_is_500(self, main_module, client, temp_user_db, monkeypatch):
+        def boom(**kwargs):
+            raise sqlite3.OperationalError("fake failure")
+
+        monkeypatch.setattr(temp_user_db, "get_user", boom)
+        with _mode(main_module, "builtin"):
+            response = _login(client)
+        assert response.status_code == 500
+        assert response.get_json() == {"error": "Authentication system error"}
+
+    def test_unexpected_mode_is_500(self, main_module, client):
+        with _mode(main_module, "unavailable"):
+            response = _login(client)
+        assert response.status_code == 500
+        assert response.get_json() == {"error": "Unknown authentication mode"}
+
+
+class TestLoginNoneMode:
+    def test_any_credentials_succeed_and_clear_failures(self, main_module, client):
+        main_module.failed_login_attempts["guest"] = {"count": 2}
+        with _mode(main_module, "none"):
+            response = _login(client, username="guest", password="any", remember_me=False)
+        assert response.status_code == 200
+        assert response.get_json() == {"success": True}
+        with client.session_transaction() as sess:
+            assert sess["user_id"] == "guest"
+            assert "is_admin" not in sess
+            assert sess.permanent is False
+        assert "guest" not in main_module.failed_login_attempts
+
+
+class TestLoginCwa:
+    @staticmethod
+    def _cwa_db(path, name="cwa_fake", password="cwahash", role=0, email=None):
+        conn = sqlite3.connect(path)
+        conn.execute(
+            "CREATE TABLE user (name TEXT PRIMARY KEY, password TEXT, role INTEGER, email TEXT)"
+        )
+        conn.execute(
+            "INSERT INTO user (name, password, role, email) VALUES (?, ?, ?, ?)",
+            (name, password, role, email),
+        )
+        conn.commit()
+        conn.close()
+        return path
+
+    def test_success_non_admin(self, main_module, client, tmp_path, temp_user_db):
+        path = self._cwa_db(tmp_path / "app.db", role=0, email="c@example.invalid")
+        with (
+            _mode(main_module, "cwa"),
+            patch.object(main_module, "CWA_DB_PATH", path),
+            patch.object(main_module, "check_password_hash", return_value=True),
+        ):
+            response = _login(client, username="cwa_fake", remember_me=True)
+        assert response.status_code == 200
+        assert response.get_json() == {"success": True}
+        with client.session_transaction() as sess:
+            assert sess["user_id"] == "cwa_fake"
+            assert sess["is_admin"] is False
+            assert sess["db_user_id"] == temp_user_db.get_user(username="cwa_fake")["id"]
+            assert sess.permanent is True
+
+    def test_admin_bit_makes_admin(self, main_module, client, tmp_path, temp_user_db):
+        path = self._cwa_db(tmp_path / "app.db", role=3)
+        with (
+            _mode(main_module, "cwa"),
+            patch.object(main_module, "CWA_DB_PATH", path),
+            patch.object(main_module, "check_password_hash", return_value=True),
+        ):
+            response = _login(client, username="cwa_fake")
+        assert response.status_code == 200
+        with client.session_transaction() as sess:
+            assert sess["is_admin"] is True
+
+    def test_wrong_password_is_counted(self, main_module, client, tmp_path, temp_user_db):
+        path = self._cwa_db(tmp_path / "app.db")
+        with (
+            _mode(main_module, "cwa"),
+            patch.object(main_module, "CWA_DB_PATH", path),
+            patch.object(main_module, "check_password_hash", return_value=False),
+        ):
+            response = _login(client, username="cwa_fake")
+        assert response.status_code == 401
+        assert response.get_json() == {"error": "Invalid username or password."}
+        assert main_module.failed_login_attempts["cwa_fake"]["count"] == 1
+
+    def test_unknown_cwa_user_is_counted(self, main_module, client, tmp_path, temp_user_db):
+        path = self._cwa_db(tmp_path / "app.db")
+        with _mode(main_module, "cwa"), patch.object(main_module, "CWA_DB_PATH", path):
+            response = _login(client, username="someone_else")
+        assert response.status_code == 401
+        assert main_module.failed_login_attempts["someone_else"]["count"] == 1
+
+    def test_missing_database_is_500(self, main_module, client, tmp_path):
+        with (
+            _mode(main_module, "cwa"),
+            patch.object(main_module, "CWA_DB_PATH", tmp_path / "gone.db"),
+        ):
+            response = _login(client, username="cwa_fake")
+        assert response.status_code == 500
+        assert response.get_json() == {"error": "Database configuration error"}
+
+    def test_unset_database_path_is_500(self, main_module, client):
+        with _mode(main_module, "cwa"), patch.object(main_module, "CWA_DB_PATH", None):
+            response = _login(client, username="cwa_fake")
+        assert response.status_code == 500
+        assert response.get_json() == {"error": "Database configuration error"}
+
+    def test_unreadable_database_is_500(self, main_module, client, tmp_path):
+        path = tmp_path / "app.db"
+        path.write_text("this is not a sqlite file")
+        with _mode(main_module, "cwa"), patch.object(main_module, "CWA_DB_PATH", path):
+            response = _login(client, username="cwa_fake")
+        assert response.status_code == 500
+        assert response.get_json() == {"error": "Authentication system error"}
+
+
+class TestLoginLockout:
+    def test_warning_starts_at_five_remaining(self, main_module, client, temp_user_db):
+        _make_alice(temp_user_db)
+        messages = []
+        with _mode(main_module, "builtin"):
+            for _ in range(9):
+                response = _login(client, password="wrong")
+                assert response.status_code == 401
+                messages.append(response.get_json()["error"])
+        assert messages[:4] == ["Invalid username or password."] * 4
+        assert messages[4] == "Invalid username or password. 5 attempts remaining."
+        assert messages[8] == "Invalid username or password. 1 attempts remaining."
+        assert main_module.failed_login_attempts["alice"]["count"] == 9
+        assert "lockout_until" not in main_module.failed_login_attempts["alice"]
+
+    def test_tenth_failure_locks_for_thirty_minutes(self, main_module, client, temp_user_db):
+        _make_alice(temp_user_db)
+        before = datetime.now(UTC)
+        with _mode(main_module, "builtin"):
+            for _ in range(9):
+                _login(client, password="wrong")
+            response = _login(client, password="wrong")
+        after = datetime.now(UTC)
+        assert response.status_code == 429
+        assert response.get_json() == {
+            "error": "Account locked due to 10 failed login attempts. Try again in 30 minutes."
+        }
+        until = main_module.failed_login_attempts["alice"]["lockout_until"]
+        assert before + timedelta(minutes=30) <= until <= after + timedelta(minutes=30)
+
+    def test_locked_account_refuses_even_the_right_password(
+        self, main_module, client, temp_user_db
+    ):
+        _make_alice(temp_user_db)
+        main_module.failed_login_attempts["alice"] = {
+            "count": 10,
+            "lockout_until": datetime.now(UTC) + timedelta(minutes=12, seconds=30),
+        }
+        with _mode(main_module, "builtin"):
+            response = _login(client)
+        assert response.status_code == 429
+        assert response.get_json() == {
+            "error": "Account temporarily locked due to multiple failed login attempts. "
+            "Try again in 12 minutes."
+        }
+        with client.session_transaction() as sess:
+            assert "user_id" not in sess
+        # A refused attempt does not extend the lockout or bump the counter.
+        assert main_module.failed_login_attempts["alice"]["count"] == 10
+
+    def test_lockout_is_per_username(self, main_module, client, temp_user_db):
+        _make_alice(temp_user_db)
+        temp_user_db.create_user(
+            username="bob", password_hash=generate_password_hash("fake-2"), role="user"
+        )
+        main_module.failed_login_attempts["alice"] = {
+            "count": 10,
+            "lockout_until": datetime.now(UTC) + timedelta(minutes=5),
+        }
+        with _mode(main_module, "builtin"):
+            response = _login(client, username="bob", password="fake-2")
+        assert response.status_code == 200
+
+    def test_lockout_applies_in_none_and_cwa_modes_too(self, main_module, client):
+        main_module.failed_login_attempts["guest"] = {
+            "count": 10,
+            "lockout_until": datetime.now(UTC) + timedelta(minutes=5),
+        }
+        with _mode(main_module, "none"):
+            response = _login(client, username="guest")
+        assert response.status_code == 429
+
+    def test_expired_lockout_lets_the_user_in_and_forgets_the_entry(
+        self, main_module, client, temp_user_db
+    ):
+        _make_alice(temp_user_db)
+        main_module.failed_login_attempts["alice"] = {
+            "count": 10,
+            "lockout_until": datetime.now(UTC) - timedelta(seconds=1),
+        }
+        with _mode(main_module, "builtin"):
+            response = _login(client)
+        assert response.status_code == 200
+        assert "alice" not in main_module.failed_login_attempts
+
+    def test_expired_lockout_restarts_the_count_after_a_failure(
+        self, main_module, client, temp_user_db
+    ):
+        _make_alice(temp_user_db)
+        main_module.failed_login_attempts["alice"] = {
+            "count": 10,
+            "lockout_until": datetime.now(UTC) - timedelta(minutes=1),
+        }
+        with _mode(main_module, "builtin"):
+            response = _login(client, password="wrong")
+        assert response.status_code == 401
+        assert main_module.failed_login_attempts["alice"] == {"count": 1}
+
+    def test_expiry_sweep_only_drops_expired_entries(self, main_module):
+        now = datetime.now(UTC)
+        main_module.failed_login_attempts.clear()
+        main_module.failed_login_attempts.update(
+            {
+                "expired": {"count": 10, "lockout_until": now - timedelta(minutes=1)},
+                "active": {"count": 10, "lockout_until": now + timedelta(minutes=1)},
+                "partial": {"count": 4},
+            }
+        )
+        try:
+            assert main_module.is_account_locked("active") is True
+            assert set(main_module.failed_login_attempts) == {"active", "partial"}
+        finally:
+            main_module.failed_login_attempts.clear()
+
+    def test_malformed_timestamp_below_threshold_is_ignored(self, main_module):
+        main_module.failed_login_attempts.clear()
+        main_module.failed_login_attempts["odd"] = {"count": 2, "lockout_until": "tomorrow"}
+        try:
+            assert main_module.is_account_locked("odd") is False
+        finally:
+            main_module.failed_login_attempts.clear()
+
+    def test_helpers_return_and_mutate_the_shared_state(self, main_module):
+        main_module.failed_login_attempts.clear()
+        state = main_module.failed_login_attempts
+        try:
+            for attempt in range(1, 10):
+                assert main_module.record_failed_login("u", "203.0.113.9") is False
+                assert state["u"]["count"] == attempt
+            assert main_module.record_failed_login("u", "203.0.113.9") is True
+            assert main_module.is_account_locked("u") is True
+            main_module.clear_failed_logins("u")
+            assert "u" not in state
+            main_module.clear_failed_logins("never-seen")
+            assert main_module.failed_login_attempts is state
+        finally:
+            main_module.failed_login_attempts.clear()
+
+    def test_limits(self, main_module):
+        assert main_module.MAX_LOGIN_ATTEMPTS == 10
+        assert main_module.LOCKOUT_DURATION_MINUTES == 30
+        assert main_module.LOGIN_ATTEMPT_WARNING_THRESHOLD == 5
+
+
+class TestLoginLocalAuthSwitches:
+    @pytest.mark.parametrize("mode", ["builtin", "oidc"])
+    def test_disable_local_auth_refuses_before_the_lockout_check(self, main_module, client, mode):
+        main_module.failed_login_attempts["alice"] = {
+            "count": 10,
+            "lockout_until": datetime.now(UTC) + timedelta(minutes=5),
+        }
+        with _mode(main_module, mode), patch.object(main_module, "DISABLE_LOCAL_AUTH", True):
+            response = _login(client)
+        assert response.status_code == 403
+        assert response.get_json() == {"error": "Local authentication is disabled"}
+
+    def test_hide_local_auth_only_blocks_oidc(self, main_module, client, temp_user_db):
+        _make_alice(temp_user_db)
+        with _mode(main_module, "builtin"), patch.object(main_module, "HIDE_LOCAL_AUTH", True):
+            assert _login(client).status_code == 200
+        with _mode(main_module, "oidc"), patch.object(main_module, "HIDE_LOCAL_AUTH", True):
+            assert _login(client).status_code == 403
+
+    def test_disable_local_auth_does_not_affect_none_mode(self, main_module, client):
+        with _mode(main_module, "none"), patch.object(main_module, "DISABLE_LOCAL_AUTH", True):
+            assert _login(client).status_code == 200
+
+
+class TestLoginClientIp:
+    def test_the_client_address_is_looked_up_per_request(self, main_module, client, temp_user_db):
+        _make_alice(temp_user_db)
+        with (
+            _mode(main_module, "builtin"),
+            patch.object(main_module, "get_client_ip", return_value="198.51.100.7") as lookup,
+        ):
+            response = _login(client, password="wrong")
+        assert response.status_code == 401
+        assert lookup.called
+
+
+class TestLogout:
+    @pytest.mark.parametrize("mode", ["builtin", "oidc", "cwa", "none"])
+    def test_clears_the_session_and_answers_success(self, main_module, client, mode):
+        with client.session_transaction() as sess:
+            sess["user_id"] = "alice"
+            sess["db_user_id"] = 1
+            sess["is_admin"] = True
+        with _mode(main_module, mode):
+            response = client.post("/api/auth/logout")
+        assert response.status_code == 200
+        assert response.get_json() == {"success": True}
+        with client.session_transaction() as sess:
+            assert dict(sess) == {}
+
+    def test_without_a_session_still_succeeds(self, main_module, client):
+        with _mode(main_module, "builtin"):
+            response = client.post("/api/auth/logout")
+        assert response.status_code == 200
+        assert response.get_json() == {"success": True}
+
+    def test_proxy_without_a_configured_url_has_no_logout_url(self, main_module, client):
+        with (
+            _mode(main_module, "proxy"),
+            patch.object(
+                main_module.app_config,
+                "get",
+                side_effect=lambda key, default=None, user_id=None: default,
+            ),
+        ):
+            response = client.post("/api/auth/logout")
+        assert response.get_json() == {"success": True}
+
+    def test_a_non_proxy_mode_ignores_the_logout_url(self, main_module, client):
+        with (
+            _mode(main_module, "builtin"),
+            patch.object(
+                main_module.app_config,
+                "get",
+                side_effect=lambda key, default=None, user_id=None: {
+                    "PROXY_AUTH_LOGOUT_URL": "https://auth.example.invalid/out"
+                }.get(key, default),
+            ),
+        ):
+            response = client.post("/api/auth/logout")
+        assert response.get_json() == {"success": True}
+
+    def test_only_post_is_allowed(self, main_module):
+        rules = [r for r in main_module.app.url_map.iter_rules() if r.rule == "/api/auth/logout"]
+        assert [sorted(r.methods - {"OPTIONS"}) for r in rules] == [["POST"]]
+
+    def test_error_is_500(self, main_module, client):
+        with (
+            _mode(main_module, "builtin"),
+            patch.object(main_module, "get_client_ip", side_effect=RuntimeError("fake")),
+        ):
+            response = client.post("/api/auth/logout")
+        assert response.status_code == 500
+        assert response.get_json() == {"error": "Logout failed"}
+
+
+class TestAuthCheck:
+    def test_none_mode_ignores_the_session(self, main_module, client):
+        with _mode(main_module, "none"):
+            response = client.get("/api/auth/check")
+        assert response.get_json() == {
+            "authenticated": True,
+            "auth_required": False,
+            "auth_mode": "none",
+            "is_admin": True,
+        }
+
+    def test_builtin_without_session(self, main_module, client, temp_user_db):
+        with _mode(main_module, "builtin"):
+            response = client.get("/api/auth/check")
+        assert response.status_code == 200
+        assert response.get_json() == {
+            "authenticated": False,
+            "auth_required": True,
+            "auth_mode": "builtin",
+            "is_admin": False,
+            "username": None,
+            "display_name": None,
+        }
+
+    def test_non_admin_session_reports_not_admin(self, main_module, client, temp_user_db):
+        user = _make_alice(temp_user_db, role="user")
+        with client.session_transaction() as sess:
+            sess["user_id"] = "alice"
+            sess["db_user_id"] = user["id"]
+            sess["is_admin"] = False
+        with _mode(main_module, "builtin"):
+            body = client.get("/api/auth/check").get_json()
+        assert body["authenticated"] is True
+        assert body["is_admin"] is False
+        assert body["username"] == "alice"
+
+    def test_cwa_session_admin_flag_comes_from_the_session(self, main_module, client):
+        with client.session_transaction() as sess:
+            sess["user_id"] = "cwa_fake"
+            sess["is_admin"] = True
+        with _mode(main_module, "cwa"):
+            body = client.get("/api/auth/check").get_json()
+        assert body == {
+            "authenticated": True,
+            "auth_required": True,
+            "auth_mode": "cwa",
+            "is_admin": True,
+            "username": "cwa_fake",
+            "display_name": None,
+        }
+
+    def test_oidc_options(self, main_module, client):
+        values = {"OIDC_BUTTON_LABEL": "Fake SSO"}
+        with (
+            _mode(main_module, "oidc"),
+            patch.object(main_module, "HIDE_LOCAL_AUTH", True),
+            patch.object(main_module, "OIDC_AUTO_REDIRECT", True),
+            patch.object(
+                main_module.app_config,
+                "get",
+                side_effect=lambda key, default=None, user_id=None: values.get(key, default),
+            ),
+        ):
+            body = client.get("/api/auth/check").get_json()
+        assert body["oidc_button_label"] == "Fake SSO"
+        assert body["hide_local_auth"] is True
+        assert body["oidc_auto_redirect"] is True
+
+    def test_oidc_without_options_has_no_extra_keys(self, main_module, client):
+        with (
+            _mode(main_module, "oidc"),
+            patch.object(main_module, "DISABLE_LOCAL_AUTH", False),
+            patch.object(main_module, "HIDE_LOCAL_AUTH", False),
+            patch.object(main_module, "OIDC_AUTO_REDIRECT", False),
+            patch.object(
+                main_module.app_config,
+                "get",
+                side_effect=lambda key, default=None, user_id=None: default,
+            ),
+        ):
+            body = client.get("/api/auth/check").get_json()
+        assert set(body) == {
+            "authenticated",
+            "auth_required",
+            "auth_mode",
+            "is_admin",
+            "username",
+            "display_name",
+        }
+
+    def test_proxy_logout_url_needs_the_user_header_setting(self, main_module, client):
+        def build(values):
+            return lambda key, default=None, user_id=None: values.get(key, default)
+
+        url = {"PROXY_AUTH_LOGOUT_URL": "https://auth.example.invalid/out"}
+        with client.session_transaction() as sess:
+            sess["user_id"] = "p"
+            sess["is_admin"] = False
+        with (
+            _mode(main_module, "proxy"),
+            patch.object(main_module.app_config, "get", side_effect=build(url)),
+        ):
+            assert "logout_url" not in client.get("/api/auth/check").get_json()
+        with (
+            _mode(main_module, "proxy"),
+            patch.object(
+                main_module.app_config,
+                "get",
+                side_effect=build({**url, "PROXY_AUTH_USER_HEADER": "X-Fake"}),
+            ),
+        ):
+            body = client.get("/api/auth/check").get_json()
+        assert body["logout_url"] == "https://auth.example.invalid/out"
+
+    def test_display_name_lookup_failure_is_ignored(
+        self, main_module, client, temp_user_db, monkeypatch
+    ):
+        user = _make_alice(temp_user_db)
+        with client.session_transaction() as sess:
+            sess["user_id"] = "alice"
+            sess["db_user_id"] = user["id"]
+            sess["is_admin"] = False
+
+        def boom(**kwargs):
+            raise sqlite3.Error("fake")
+
+        monkeypatch.setattr(temp_user_db, "get_user", boom)
+        with _mode(main_module, "builtin"):
+            body = client.get("/api/auth/check").get_json()
+        assert body["authenticated"] is True
+        assert body["display_name"] is None
+
+    def test_error_reports_unknown_mode(self, main_module, client):
+        with (
+            _mode(main_module, "builtin"),
+            patch.object(
+                main_module, "get_auth_check_admin_status", side_effect=RuntimeError("fake")
+            ),
+        ):
+            response = client.get("/api/auth/check")
+        assert response.status_code == 200
+        assert response.get_json() == {
+            "authenticated": False,
+            "auth_required": True,
+            "auth_mode": "unknown",
+            "is_admin": False,
+        }
+
+    def test_only_get_is_allowed(self, main_module):
+        rules = [r for r in main_module.app.url_map.iter_rules() if r.rule == "/api/auth/check"]
+        assert [sorted(r.methods - {"OPTIONS"}) for r in rules] == [["GET", "HEAD"]]