refactor: move the proxy-auth and API-key middleware into core/auth_middleware.py (#1641)

Row 8 of the plan in Discussion #1536 is the auth middleware. The API
key `before_request` hook, the proxy auth hook, the hook that strips the
session cookie from API key requests and `_proxy_default_is_admin` move
from `main.py` to `core/auth_middleware.py`, registered with
`register_auth_middleware(app, resolve_auth_mode=..., get_user_db=...)`.

This is the code that decides who a request is, so the first commit adds
41 tests before anything moves. They cover the hook order, the admin
group mapping (comma and pipe delimiters, a custom group header, the
WSGI fallback), proxy header handling in each auth mode, provisioning
once per session, the API key scope (exempt paths, a read only key only
on GET `/api/stats`, a wrong key) and the cookie stripping. They pass on
current `main`. The second commit is the move.

The call is made right after `register_security_headers`, so the
`before_request` and `after_request` order is the same as before. The
existing tests called `proxy_auth_middleware()` and
`api_key_auth_middleware()` on `shelfmark.main` directly, and those
names are gone, so 18 call lines in three test files now find the
registered hook by name through a small helper. No patch targets changed
since the auth mode and user database are still read from `main` at
request time.

The URL map and hook order are identical, and the full suite, ruff and
basedpyright are clean.


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

<!-- coderabbit-ai-generated:v1 -->

## Summary by CodeRabbit

* **Authentication**
* API-key authentication supports admin and read-only access, with
read-only keys limited to statistics requests.
* Reverse-proxy authentication recognizes configured identity and group
information, and applies admin access according to the configured role
settings.
* Requests with invalid credentials, missing proxy identities, or
authentication errors receive appropriate error responses.
  * API-key requests no longer retain session cookies.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

shelfmark/core/auth_middleware.py

@@ -0,0 +1,278 @@
+"""Request hooks for the two ways a request can be authenticated without logging in: the
+SHELFMARK_API_KEY credentials and the reverse proxy's identity headers.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from typing import TYPE_CHECKING, NoReturn
+
+from flask import Response, g, jsonify, request, session
+
+from shelfmark.core import api_key as api_key_module  # module access lets tests monkeypatch the key
+from shelfmark.core.config import config as app_config
+from shelfmark.core.external_user_linking import upsert_external_user
+from shelfmark.core.logger import setup_logger
+from shelfmark.core.request_helpers import normalize_optional_text
+from shelfmark.core.stats_routes import STATS_PATH
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from flask import Flask
+
+    from shelfmark.core.user_db import UserDB
+
+logger = setup_logger(__name__)
+
+_OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error)
+
+
+def _raise_runtime_error(message: str) -> NoReturn:
+    raise RuntimeError(message)
+
+
+def _proxy_default_is_admin(db: UserDB) -> bool:
+    """Role for a proxy user seen for the first time when no admin group is configured.
+
+    The first account ever provisioned is an admin so the instance is never left without
+    one; later accounts follow PROXY_AUTH_DEFAULT_ROLE (default: user).
+    """
+    if not db.has_admin():
+        return True
+    role = str(app_config.get("PROXY_AUTH_DEFAULT_ROLE", "user") or "user").strip().lower()
+    return role == "admin"
+
+
+_API_KEY_EXEMPT_PREFIXES = ("/api/auth/",)
+_API_KEY_EXEMPT_PATHS = frozenset({"/api/health"})
+_READ_ONLY_KEY_PATHS = frozenset({STATS_PATH})
+
+
+def register_auth_middleware(
+    app: Flask,
+    *,
+    resolve_auth_mode: Callable[[], str],
+    get_user_db: Callable[[], UserDB | None],
+) -> None:
+    """Register the API-key and proxy-auth ``before_request`` hooks and the cookie-stripping
+    ``after_request`` hook.
+
+    Call this at the place the hooks should take in the hook order: ``before_request`` hooks run
+    in registration order, ``after_request`` hooks in reverse. The API-key hook must come first
+    because the proxy hook skips requests it already authenticated.
+    """
+
+    @app.before_request
+    def api_key_auth_middleware() -> Response | tuple[Response, int] | None:
+        """Authenticate requests that present the configured SHELFMARK_API_KEY.
+
+        Both Authorization: Bearer and X-Api-Key are checked, and either
+        matching authenticates the request as an admin for this request only:
+        any session cookie is ignored and none is written back. Checking both
+        means a reverse proxy's own Authorization header never shadows an
+        operator-supplied X-Api-Key. No matching candidate is ignored so bearer
+        tokens forwarded by reverse proxies keep working; the request then
+        continues on the normal session path.
+        """
+        if not request.path.startswith("/api/"):
+            return None
+        if request.path in _API_KEY_EXEMPT_PATHS or request.path.startswith(
+            _API_KEY_EXEMPT_PREFIXES
+        ):
+            return None
+        if not api_key_module.SHELFMARK_API_KEY and not api_key_module.SHELFMARK_API_KEY_READONLY:
+            return None
+
+        candidates = api_key_module.extract_api_key_candidates(
+            request.headers.get("Authorization"), request.headers.get("X-Api-Key")
+        )
+        if not candidates:
+            return None
+
+        scopes = {api_key_module.key_scope(candidate) for candidate in candidates}
+        scope = "admin" if "admin" in scopes else "readonly" if "readonly" in scopes else None
+        if scope is None:
+            return None
+        if resolve_auth_mode() == "none":
+            return None
+
+        # Mark the request as keyed before the lookup so the after-request cookie
+        # reset also covers the error path below.
+        g.api_key_auth = True
+
+        if scope == "readonly":
+            # A read-only key is a dashboard credential: one GET endpoint, no session.
+            g.api_key_scope = "readonly"
+            if request.method != "GET" or request.path not in _READ_ONLY_KEY_PATHS:
+                return jsonify({"error": "This API key is read-only"}), 403
+            return None
+        g.api_key_scope = "admin"
+
+        try:
+            user_db = get_user_db()
+            admin = user_db.get_first_admin() if user_db is not None else None
+        except _OPERATIONAL_ERRORS:
+            logger.exception("API key auth middleware error")
+            return jsonify({"error": "Authentication error"}), 500
+
+        session.clear()
+        session["user_id"] = admin["username"] if admin else "api"
+        session["is_admin"] = True
+        if admin:
+            session["db_user_id"] = admin["id"]
+        session.permanent = False
+        # Identity is per request; never persist it as a cookie.
+        session.modified = False
+        return None
+
+    @app.before_request
+    def proxy_auth_middleware() -> Response | tuple[Response, int] | None:
+        """Middleware to handle proxy authentication.
+
+        When AUTH_METHOD is set to "proxy", this middleware automatically
+        authenticates users based on headers set by the reverse proxy.
+        """
+        auth_mode = resolve_auth_mode()
+
+        # Only run for proxy auth mode
+        if auth_mode != "proxy":
+            return None
+
+        # A request already authenticated by SHELFMARK_API_KEY needs no proxy headers.
+        if g.get("api_key_auth"):
+            return None
+
+        # Skip for public endpoints that don't need auth
+        if request.path == "/api/health":
+            return None
+
+        def get_proxy_header(header_name: str) -> str | None:
+            """Resolve proxy auth values from headers with WSGI env fallbacks."""
+            value = request.headers.get(header_name)
+            if value:
+                return value
+
+            env_key = f"HTTP_{header_name.upper().replace('-', '_')}"
+            value = request.environ.get(env_key)
+            if value:
+                return value
+
+            # Some proxies set authenticated username in REMOTE_USER (not as a header).
+            if header_name.lower().replace("_", "-") == "remote-user":
+                return request.environ.get("REMOTE_USER")
+
+            return None
+
+        user_db = get_user_db()
+
+        try:
+            user_header = (
+                normalize_optional_text(app_config.get("PROXY_AUTH_USER_HEADER", "X-Auth-User"))
+                or "X-Auth-User"
+            )
+
+            # Extract username from proxy header
+            username = get_proxy_header(user_header)
+
+            if not username:
+                if request.path.startswith("/api/auth/"):
+                    return None
+
+                logger.warning(
+                    "Proxy auth enabled but no username found in header '%s'", user_header
+                )
+                return jsonify({"error": "Authentication required. Proxy header not set."}), 401
+
+            # Resolve admin role for proxy sessions.
+            # If an admin group is configured, derive from groups header.
+            # Otherwise preserve the existing DB role for known users; a first-time user
+            # is an admin only while the instance has none (so nobody is locked out),
+            # after that PROXY_AUTH_DEFAULT_ROLE decides (default: user).
+            admin_group_header = (
+                normalize_optional_text(
+                    app_config.get("PROXY_AUTH_ADMIN_GROUP_HEADER", "X-Auth-Groups")
+                )
+                or "X-Auth-Groups"
+            )
+            admin_group_name = (
+                normalize_optional_text(app_config.get("PROXY_AUTH_ADMIN_GROUP_NAME", "")) or ""
+            )
+            is_admin = True
+
+            if admin_group_name:
+                groups_header = get_proxy_header(admin_group_header) or ""
+                user_groups_delimiter = "," if "," in groups_header else "|"
+                user_groups = [
+                    g.strip() for g in groups_header.split(user_groups_delimiter) if g.strip()
+                ]
+                is_admin = admin_group_name in user_groups
+            elif user_db is not None:
+                existing_db_user = user_db.get_user_cached(username=username)
+                if existing_db_user:
+                    is_admin = existing_db_user.get("role") == "admin"
+                else:
+                    is_admin = _proxy_default_is_admin(user_db)
+
+            # Create or update session
+            previous_username = session.get("user_id")
+            if previous_username and previous_username != username:
+                # Header identity changed mid-session; force reprovision for the new user.
+                session.pop("db_user_id", None)
+
+            session["user_id"] = username
+            session["is_admin"] = is_admin
+
+            # Provision proxy-authenticated users into users.db for multi-user features.
+            # Re-provision when db_user_id is missing/stale/mismatched to avoid broken
+            # sessions after DB resets or auth-mode transitions.
+            if user_db is not None:
+                raw_db_user_id = session.get("db_user_id")
+                session_db_user = None
+
+                if raw_db_user_id is not None:
+                    try:
+                        session_db_user = user_db.get_user_cached(user_id=int(raw_db_user_id))
+                    except TypeError, ValueError:
+                        session_db_user = None
+
+                session_db_username = (
+                    str(session_db_user.get("username") or "").strip() if session_db_user else ""
+                )
+                needs_db_user_sync = (
+                    raw_db_user_id is None
+                    or session_db_user is None
+                    or session_db_username != username
+                )
+
+                if needs_db_user_sync:
+                    role = "admin" if is_admin else "user"
+                    db_user, _ = upsert_external_user(
+                        user_db,
+                        auth_source="proxy",
+                        username=username,
+                        role=role,
+                        collision_strategy="takeover",
+                        context="proxy_request",
+                    )
+                    if db_user is None:
+                        _raise_runtime_error("Unexpected proxy user sync result: no user returned")
+
+                    session["db_user_id"] = db_user["id"]
+
+            session.permanent = False
+        except _OPERATIONAL_ERRORS:
+            logger.exception("Proxy auth middleware error")
+            return jsonify({"error": "Authentication error"}), 500
+        else:
+            return None
+
+    @app.after_request
+    def strip_cookie_for_api_key_requests(response: Response) -> Response:
+        """Keyed requests never mint or refresh a session cookie, even if a handler dirties the session."""
+        if g.get("api_key_auth"):
+            # Setting `permanent` mutates the session dict (re-marking it
+            # modified), so it must be reset before `modified`, not after.
+            session.permanent = False
+            session.modified = False
+        return response

shelfmark/main.py

@@ -9,7 +9,7 @@ from importlib import import_module
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, NoReturn, cast
 
-from flask import Flask, g, jsonify, request, session
+from flask import Flask, jsonify, request, session
 from flask_cors import CORS
 from flask_socketio import SocketIO
 from werkzeug.middleware.proxy_fix import ProxyFix
@@ -34,7 +34,6 @@ from shelfmark.config.security import _migrate_security_settings
 from shelfmark.config.settings import (
     migrate_audiobook_format_settings,
 )
-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 (
@@ -42,6 +41,7 @@ from shelfmark.core.app_setup import (
     configure_secret_key_and_session,
     register_security_headers,
 )
+from shelfmark.core.auth_middleware import register_auth_middleware
 from shelfmark.core.auth_modes import (
     get_auth_check_admin_status,
     is_settings_or_onboarding_path,
@@ -62,7 +62,6 @@ from shelfmark.core.auth_routes import (  # noqa: F401
 )
 from shelfmark.core.config import config as app_config
 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
 from shelfmark.core.metadata_routes import register_metadata_routes
 from shelfmark.core.models import (
@@ -104,7 +103,7 @@ from shelfmark.core.requests_service import (
 from shelfmark.core.settings_routes import register_settings_routes
 from shelfmark.core.socket_handlers import register_socket_handlers
 from shelfmark.core.socketio_origins import build_cors_allowed_origins
-from shelfmark.core.stats_routes import STATS_PATH, register_stats_routes
+from shelfmark.core.stats_routes import register_stats_routes
 from shelfmark.core.system_routes import DebugHooks, register_system_routes
 from shelfmark.core.user_db import UserDB
 from shelfmark.core.utils import AUDIOBOOK_FORMATS, normalize_base_path
@@ -652,232 +651,12 @@ logger.info(
 logger.info("Session cookie name: %s", SESSION_COOKIE_NAME)
 
 
-def _proxy_default_is_admin(db: UserDB) -> bool:
-    """Role for a proxy user seen for the first time when no admin group is configured.
-
-    The first account ever provisioned is an admin so the instance is never left without
-    one; later accounts follow PROXY_AUTH_DEFAULT_ROLE (default: user).
-    """
-    if not db.has_admin():
-        return True
-    role = str(app_config.get("PROXY_AUTH_DEFAULT_ROLE", "user") or "user").strip().lower()
-    return role == "admin"
-
-
-_API_KEY_EXEMPT_PREFIXES = ("/api/auth/",)
-_API_KEY_EXEMPT_PATHS = frozenset({"/api/health"})
-_READ_ONLY_KEY_PATHS = frozenset({STATS_PATH})
-
-
-@app.before_request
-def api_key_auth_middleware() -> Response | tuple[Response, int] | None:
-    """Authenticate requests that present the configured SHELFMARK_API_KEY.
-
-    Both Authorization: Bearer and X-Api-Key are checked, and either
-    matching authenticates the request as an admin for this request only:
-    any session cookie is ignored and none is written back. Checking both
-    means a reverse proxy's own Authorization header never shadows an
-    operator-supplied X-Api-Key. No matching candidate is ignored so bearer
-    tokens forwarded by reverse proxies keep working; the request then
-    continues on the normal session path.
-    """
-    if not request.path.startswith("/api/"):
-        return None
-    if request.path in _API_KEY_EXEMPT_PATHS or request.path.startswith(_API_KEY_EXEMPT_PREFIXES):
-        return None
-    if not api_key_module.SHELFMARK_API_KEY and not api_key_module.SHELFMARK_API_KEY_READONLY:
-        return None
-
-    candidates = api_key_module.extract_api_key_candidates(
-        request.headers.get("Authorization"), request.headers.get("X-Api-Key")
-    )
-    if not candidates:
-        return None
-
-    scopes = {api_key_module.key_scope(candidate) for candidate in candidates}
-    scope = "admin" if "admin" in scopes else "readonly" if "readonly" in scopes else None
-    if scope is None:
-        return None
-    if get_auth_mode() == "none":
-        return None
-
-    # Mark the request as keyed before the lookup so the after-request cookie
-    # reset also covers the error path below.
-    g.api_key_auth = True
-
-    if scope == "readonly":
-        # A read-only key is a dashboard credential: one GET endpoint, no session.
-        g.api_key_scope = "readonly"
-        if request.method != "GET" or request.path not in _READ_ONLY_KEY_PATHS:
-            return jsonify({"error": "This API key is read-only"}), 403
-        return None
-    g.api_key_scope = "admin"
-
-    try:
-        admin = user_db.get_first_admin() if user_db is not None else None
-    except _OPERATIONAL_ERRORS:
-        logger.exception("API key auth middleware error")
-        return jsonify({"error": "Authentication error"}), 500
-
-    session.clear()
-    session["user_id"] = admin["username"] if admin else "api"
-    session["is_admin"] = True
-    if admin:
-        session["db_user_id"] = admin["id"]
-    session.permanent = False
-    # Identity is per request; never persist it as a cookie.
-    session.modified = False
-    return None
-
-
-@app.before_request
-def proxy_auth_middleware() -> Response | tuple[Response, int] | None:
-    """Middleware to handle proxy authentication.
-
-    When AUTH_METHOD is set to "proxy", this middleware automatically
-    authenticates users based on headers set by the reverse proxy.
-    """
-    auth_mode = get_auth_mode()
-
-    # Only run for proxy auth mode
-    if auth_mode != "proxy":
-        return None
-
-    # A request already authenticated by SHELFMARK_API_KEY needs no proxy headers.
-    if g.get("api_key_auth"):
-        return None
-
-    # Skip for public endpoints that don't need auth
-    if request.path == "/api/health":
-        return None
-
-    def get_proxy_header(header_name: str) -> str | None:
-        """Resolve proxy auth values from headers with WSGI env fallbacks."""
-        value = request.headers.get(header_name)
-        if value:
-            return value
-
-        env_key = f"HTTP_{header_name.upper().replace('-', '_')}"
-        value = request.environ.get(env_key)
-        if value:
-            return value
-
-        # Some proxies set authenticated username in REMOTE_USER (not as a header).
-        if header_name.lower().replace("_", "-") == "remote-user":
-            return request.environ.get("REMOTE_USER")
-
-        return None
-
-    try:
-        user_header = (
-            normalize_optional_text(app_config.get("PROXY_AUTH_USER_HEADER", "X-Auth-User"))
-            or "X-Auth-User"
-        )
-
-        # Extract username from proxy header
-        username = get_proxy_header(user_header)
-
-        if not username:
-            if request.path.startswith("/api/auth/"):
-                return None
-
-            logger.warning("Proxy auth enabled but no username found in header '%s'", user_header)
-            return jsonify({"error": "Authentication required. Proxy header not set."}), 401
-
-        # Resolve admin role for proxy sessions.
-        # If an admin group is configured, derive from groups header.
-        # Otherwise preserve the existing DB role for known users; a first-time user
-        # is an admin only while the instance has none (so nobody is locked out),
-        # after that PROXY_AUTH_DEFAULT_ROLE decides (default: user).
-        admin_group_header = (
-            normalize_optional_text(
-                app_config.get("PROXY_AUTH_ADMIN_GROUP_HEADER", "X-Auth-Groups")
-            )
-            or "X-Auth-Groups"
-        )
-        admin_group_name = (
-            normalize_optional_text(app_config.get("PROXY_AUTH_ADMIN_GROUP_NAME", "")) or ""
-        )
-        is_admin = True
-
-        if admin_group_name:
-            groups_header = get_proxy_header(admin_group_header) or ""
-            user_groups_delimiter = "," if "," in groups_header else "|"
-            user_groups = [
-                g.strip() for g in groups_header.split(user_groups_delimiter) if g.strip()
-            ]
-            is_admin = admin_group_name in user_groups
-        elif user_db is not None:
-            existing_db_user = user_db.get_user_cached(username=username)
-            if existing_db_user:
-                is_admin = existing_db_user.get("role") == "admin"
-            else:
-                is_admin = _proxy_default_is_admin(user_db)
-
-        # Create or update session
-        previous_username = session.get("user_id")
-        if previous_username and previous_username != username:
-            # Header identity changed mid-session; force reprovision for the new user.
-            session.pop("db_user_id", None)
-
-        session["user_id"] = username
-        session["is_admin"] = is_admin
-
-        # Provision proxy-authenticated users into users.db for multi-user features.
-        # Re-provision when db_user_id is missing/stale/mismatched to avoid broken
-        # sessions after DB resets or auth-mode transitions.
-        if user_db is not None:
-            raw_db_user_id = session.get("db_user_id")
-            session_db_user = None
-
-            if raw_db_user_id is not None:
-                try:
-                    session_db_user = user_db.get_user_cached(user_id=int(raw_db_user_id))
-                except TypeError, ValueError:
-                    session_db_user = None
-
-            session_db_username = (
-                str(session_db_user.get("username") or "").strip() if session_db_user else ""
-            )
-            needs_db_user_sync = (
-                raw_db_user_id is None or session_db_user is None or session_db_username != username
-            )
-
-            if needs_db_user_sync:
-                role = "admin" if is_admin else "user"
-                db_user, _ = upsert_external_user(
-                    user_db,
-                    auth_source="proxy",
-                    username=username,
-                    role=role,
-                    collision_strategy="takeover",
-                    context="proxy_request",
-                )
-                if db_user is None:
-                    _raise_runtime_error("Unexpected proxy user sync result: no user returned")
-
-                session["db_user_id"] = db_user["id"]
-
-        session.permanent = False
-    except _OPERATIONAL_ERRORS:
-        logger.exception("Proxy auth middleware error")
-        return jsonify({"error": "Authentication error"}), 500
-    else:
-        return None
-
-
 register_security_headers(app)
-
-
-@app.after_request
-def strip_cookie_for_api_key_requests(response: Response) -> Response:
-    """Keyed requests never mint or refresh a session cookie, even if a handler dirties the session."""
-    if g.get("api_key_auth"):
-        # Setting `permanent` mutates the session dict (re-marking it
-        # modified), so it must be reset before `modified`, not after.
-        session.permanent = False
-        session.modified = False
-    return response
+register_auth_middleware(
+    app,
+    resolve_auth_mode=lambda: get_auth_mode(),  # noqa: PLW0108
+    get_user_db=lambda: user_db,
+)
 
 
 def login_required(

tests/core/test_api_key_env.py

@@ -14,6 +14,11 @@ import pytest
 from shelfmark.core import api_key
 
 
+def _hook(main_module, name):
+    """The registered before_request hook called `name` (it lives in core/auth_middleware.py)."""
+    return next(f for f in main_module.app.before_request_funcs[None] if f.__name__ == name)
+
+
 @pytest.fixture
 def user_db():
     from shelfmark.core.user_db import UserDB
@@ -111,7 +116,7 @@ def _identity_for_key(wired):
     the middleware in a request context sidesteps that entirely.
     """
     with wired.app.test_request_context("/api/downloads/active", headers=_bearer("s3cret")):
-        assert wired.api_key_auth_middleware() is None
+        assert _hook(wired, "api_key_auth_middleware")() is None
         from flask import session
 
         return dict(session)
@@ -492,7 +497,7 @@ class TestReadOnlyKeyRequests:
     def test_never_gets_an_admin_session(self, ro, user_db):
         user_db.create_user(username="root", role="admin")
         with ro.app.test_request_context("/api/stats", headers=_bearer("ro-key")):
-            assert ro.api_key_auth_middleware() is None
+            assert _hook(ro, "api_key_auth_middleware")() is None
             from flask import session
 
             assert "is_admin" not in session

tests/core/test_auth_api.py

@@ -10,6 +10,11 @@ import pytest
 from werkzeug.security import generate_password_hash
 
 
+def _hook(main_module, name):
+    """The registered before_request hook called `name` (it lives in core/auth_middleware.py)."""
+    return next(f for f in main_module.app.before_request_funcs[None] if f.__name__ == name)
+
+
 @pytest.fixture(scope="module")
 def main_module():
     """Import `shelfmark.main` with background startup disabled."""
@@ -309,7 +314,7 @@ class TestProxyProvisioningRole:
             ):
                 main_module.session["user_id"] = "bob"
                 main_module.session["db_user_id"] = bob["id"]
-                assert main_module.proxy_auth_middleware() is None
+                assert _hook(main_module, "proxy_auth_middleware")() is None
                 return main_module.session["is_admin"]
 
         assert run_middleware() is False

tests/core/test_auth_middleware.py

@@ -0,0 +1,324 @@
+"""Pin the API-key and proxy-auth request hooks registered on the Flask app."""
+
+from __future__ import annotations
+
+import importlib
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+from flask import Response, g, session
+
+from shelfmark.core import api_key
+
+
+@pytest.fixture(scope="module")
+def main_module():
+    with patch("shelfmark.download.orchestrator.start"):
+        import shelfmark.main as main
+
+        importlib.reload(main)
+        return main
+
+
+@pytest.fixture(autouse=True)
+def _empty_user_table(main_module):
+    def _clear() -> None:
+        for user in main_module.user_db.list_users():
+            main_module.user_db.delete_user(user["id"])
+
+    _clear()
+    yield
+    _clear()
+
+
+def _hook(app, kind: str, name: str):
+    funcs = app.before_request_funcs if kind == "before" else app.after_request_funcs
+    for func in funcs[None]:
+        if func.__name__ == name:
+            return func
+    raise AssertionError(f"{kind}_request hook {name} is not registered")
+
+
+def _config_getter(values: dict[str, Any]):
+    def _get(key: str, default: Any = None, user_id: Any = None):
+        return values.get(key, default)
+
+    return _get
+
+
+def _proxy(main_module, path="/api/releases", *, headers=None, environ=None, config=None):
+    """Run the proxy hook in a request context and return (result, session dict)."""
+    values = {"PROXY_AUTH_USER_HEADER": "X-Auth-User", **(config or {})}
+    with (
+        patch.object(main_module, "get_auth_mode", return_value="proxy"),
+        patch.object(main_module.app_config, "get", side_effect=_config_getter(values)),
+        main_module.app.test_request_context(path, headers=headers, environ_base=environ),
+    ):
+        result = _hook(main_module.app, "before", "proxy_auth_middleware")()
+        return result, dict(session), session.permanent
+
+
+class TestHookOrder:
+    def test_before_request_order(self, main_module):
+        names = [f.__name__ for f in main_module.app.before_request_funcs[None]]
+        assert names.index("api_key_auth_middleware") < names.index("proxy_auth_middleware")
+
+    def test_after_request_order(self, main_module):
+        names = [f.__name__ for f in main_module.app.after_request_funcs[None]]
+        assert names.index("set_security_headers") < names.index(
+            "strip_cookie_for_api_key_requests"
+        )
+
+
+class TestProxyGroups:
+    @pytest.mark.parametrize(
+        ("groups", "expected"),
+        [
+            ("admins", True),
+            ("users,admins", True),
+            ("users, admins ,other", True),
+            ("users|admins", True),
+            ("users", False),
+            ("", False),
+            ("administrators", False),
+        ],
+    )
+    def test_group_header_maps_to_admin(self, main_module, groups, expected):
+        headers = {"X-Auth-User": "gina"}
+        if groups:
+            headers["X-Auth-Groups"] = groups
+        result, sess, permanent = _proxy(
+            main_module, headers=headers, config={"PROXY_AUTH_ADMIN_GROUP_NAME": "admins"}
+        )
+        assert result is None
+        assert sess["is_admin"] is expected
+        assert permanent is False
+        db_user = main_module.user_db.get_user(user_id=sess["db_user_id"])
+        assert db_user["role"] == ("admin" if expected else "user")
+
+    def test_custom_group_header(self, main_module):
+        result, sess, _ = _proxy(
+            main_module,
+            headers={"X-Auth-User": "gina", "Remote-Groups": "ops"},
+            config={
+                "PROXY_AUTH_ADMIN_GROUP_NAME": "ops",
+                "PROXY_AUTH_ADMIN_GROUP_HEADER": "Remote-Groups",
+            },
+        )
+        assert result is None
+        assert sess["is_admin"] is True
+
+    def test_group_header_wsgi_environ_fallback(self, main_module):
+        result, sess, _ = _proxy(
+            main_module,
+            headers={"X-Auth-User": "gina"},
+            environ={"HTTP_X_AUTH_GROUPS": "admins"},
+            config={"PROXY_AUTH_ADMIN_GROUP_NAME": "admins"},
+        )
+        assert result is None
+        assert sess["is_admin"] is True
+
+    def test_configured_group_overrides_stored_role(self, main_module):
+        main_module.user_db.create_user(username="gina", role="admin", auth_source="proxy")
+        result, sess, _ = _proxy(
+            main_module,
+            headers={"X-Auth-User": "gina", "X-Auth-Groups": "users"},
+            config={"PROXY_AUTH_ADMIN_GROUP_NAME": "admins"},
+        )
+        assert result is None
+        assert sess["is_admin"] is False
+
+    def test_blank_user_header_setting_falls_back_to_default(self, main_module):
+        result, sess, _ = _proxy(
+            main_module,
+            headers={"X-Auth-User": "gina"},
+            config={"PROXY_AUTH_USER_HEADER": "   "},
+        )
+        assert result is None
+        assert sess["user_id"] == "gina"
+
+
+class TestProxyHeaderHandling:
+    def test_missing_header_on_protected_path_is_401(self, main_module):
+        result, sess, _ = _proxy(main_module)
+        resp, status = result
+        assert status == 401
+        assert resp.get_json() == {"error": "Authentication required. Proxy header not set."}
+        assert "user_id" not in sess
+
+    def test_missing_header_on_auth_path_passes(self, main_module):
+        result, sess, _ = _proxy(main_module, "/api/auth/login")
+        assert result is None
+        assert "user_id" not in sess
+
+    def test_non_api_path_without_header_is_401(self, main_module):
+        result, _, _ = _proxy(main_module, "/")
+        assert result[1] == 401
+
+    def test_header_ignored_outside_proxy_mode(self, main_module):
+        for mode in ("builtin", "none", "cwa"):
+            with (
+                patch.object(main_module, "get_auth_mode", return_value=mode),
+                main_module.app.test_request_context(
+                    "/api/releases", headers={"X-Auth-User": "mallory"}
+                ),
+            ):
+                assert _hook(main_module.app, "before", "proxy_auth_middleware")() is None
+                assert "user_id" not in session
+        assert main_module.user_db.get_user(username="mallory") is None
+
+    def test_custom_user_header_is_case_insensitive(self, main_module):
+        result, sess, _ = _proxy(
+            main_module,
+            headers={"x-forwarded-user": "gina"},
+            config={"PROXY_AUTH_USER_HEADER": "X-Forwarded-User"},
+        )
+        assert result is None
+        assert sess["user_id"] == "gina"
+
+    def test_provisioned_once_per_session(self, main_module):
+        _, sess, _ = _proxy(main_module, headers={"X-Auth-User": "gina"})
+        first_id = sess["db_user_id"]
+        with (
+            patch.object(main_module, "get_auth_mode", return_value="proxy"),
+            main_module.app.test_request_context("/api/releases", headers={"X-Auth-User": "gina"}),
+        ):
+            session["user_id"] = "gina"
+            session["db_user_id"] = first_id
+            assert _hook(main_module.app, "before", "proxy_auth_middleware")() is None
+            assert session["db_user_id"] == first_id
+        assert len(main_module.user_db.list_users()) == 1
+
+    def test_store_error_is_500(self, main_module):
+        with patch.object(main_module.user_db, "get_user_cached", side_effect=RuntimeError("boom")):
+            result, _, _ = _proxy(main_module, headers={"X-Auth-User": "gina"})
+        resp, status = result
+        assert status == 500
+        assert resp.get_json() == {"error": "Authentication error"}
+
+
+class TestApiKeyScope:
+    @pytest.fixture
+    def keyed(self, main_module, monkeypatch):
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY", "s3cret")
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY_READONLY", "ro-key")
+        with patch.object(main_module, "get_auth_mode", return_value="builtin"):
+            yield main_module
+
+    def _run(self, main_module, path, key="s3cret", method="GET"):
+        with main_module.app.test_request_context(path, method=method, headers={"X-Api-Key": key}):
+            result = _hook(main_module.app, "before", "api_key_auth_middleware")()
+            return result, dict(session), g.get("api_key_auth"), g.get("api_key_scope")
+
+    @pytest.mark.parametrize("path", ["/", "/index.html", "/static/app.js", "/apix", "/api"])
+    def test_non_api_paths_ignore_key(self, keyed, path):
+        result, sess, flag, scope = self._run(keyed, path)
+        assert result is None
+        assert sess == {}
+        assert flag is None
+        assert scope is None
+
+    @pytest.mark.parametrize("path", ["/api/health", "/api/auth/check", "/api/auth/login"])
+    def test_exempt_paths_ignore_key(self, keyed, path):
+        result, sess, flag, _ = self._run(keyed, path)
+        assert result is None
+        assert sess == {}
+        assert flag is None
+
+    def test_admin_key_marks_request_and_sets_admin_session(self, keyed):
+        admin = keyed.user_db.create_user(username="root", role="admin")
+        result, sess, flag, scope = self._run(keyed, "/api/downloads/active")
+        assert result is None
+        assert flag is True
+        assert scope == "admin"
+        assert sess["user_id"] == "root"
+        assert sess["is_admin"] is True
+        assert sess["db_user_id"] == admin["id"]
+
+    def test_admin_key_without_admin_account_uses_api_identity(self, keyed):
+        _, sess, _, _ = self._run(keyed, "/api/downloads/active")
+        assert sess["user_id"] == "api"
+        assert sess["is_admin"] is True
+        assert "db_user_id" not in sess
+
+    def test_admin_key_session_is_not_persisted(self, keyed):
+        with keyed.app.test_request_context(
+            "/api/downloads/active", headers={"X-Api-Key": "s3cret"}
+        ):
+            _hook(keyed.app, "before", "api_key_auth_middleware")()
+            assert session.permanent is False
+            assert session.modified is False
+
+    def test_wrong_key_falls_through_untouched(self, keyed):
+        result, sess, flag, scope = self._run(keyed, "/api/downloads/active", key="nope")
+        assert result is None
+        assert sess == {}
+        assert flag is None
+        assert scope is None
+
+    def test_readonly_key_allowed_on_stats_get_only(self, keyed):
+        result, sess, flag, scope = self._run(keyed, "/api/stats", key="ro-key")
+        assert result is None
+        assert (sess, flag, scope) == ({}, True, "readonly")
+
+    @pytest.mark.parametrize(
+        ("path", "method"),
+        [("/api/stats", "POST"), ("/api/downloads/active", "GET"), ("/api/stats/x", "GET")],
+    )
+    def test_readonly_key_refused_elsewhere(self, keyed, path, method):
+        result, _, flag, scope = self._run(keyed, path, key="ro-key", method=method)
+        resp, status = result
+        assert status == 403
+        assert resp.get_json() == {"error": "This API key is read-only"}
+        assert flag is True
+        assert scope == "readonly"
+
+    def test_none_mode_ignores_key(self, main_module, monkeypatch):
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY", "s3cret")
+        with patch.object(main_module, "get_auth_mode", return_value="none"):
+            result, sess, flag, _ = self._run(main_module, "/api/downloads/active")
+        assert (result, sess, flag) == (None, {}, None)
+
+    def test_store_error_is_500_and_still_marked_keyed(self, keyed):
+        with patch.object(keyed.user_db, "get_first_admin", side_effect=RuntimeError("boom")):
+            result, _, flag, _ = self._run(keyed, "/api/downloads/active")
+        resp, status = result
+        assert status == 500
+        assert resp.get_json() == {"error": "Authentication error"}
+        assert flag is True
+
+    def test_proxy_mode_keyed_request_skips_proxy_hook(self, keyed):
+        with (
+            patch.object(keyed, "get_auth_mode", return_value="proxy"),
+            keyed.app.test_request_context(
+                "/api/downloads/active", headers={"X-Api-Key": "s3cret"}
+            ),
+        ):
+            assert _hook(keyed.app, "before", "api_key_auth_middleware")() is None
+            assert _hook(keyed.app, "before", "proxy_auth_middleware")() is None
+            assert session["is_admin"] is True
+
+
+class TestCookieStripping:
+    def _strip(self, main_module, *, keyed: bool):
+        with main_module.app.test_request_context("/api/downloads/active"):
+            if keyed:
+                g.api_key_auth = True
+            session["dirty"] = True
+            session.permanent = True
+            resp = _hook(main_module.app, "after", "strip_cookie_for_api_key_requests")(
+                Response("ok")
+            )
+            return resp, session.modified, session.permanent
+
+    def test_keyed_request_resets_session_flags(self, main_module):
+        resp, modified, permanent = self._strip(main_module, keyed=True)
+        assert resp.get_data(as_text=True) == "ok"
+        assert modified is False
+        assert permanent is False
+
+    def test_unkeyed_request_is_untouched(self, main_module):
+        _, modified, permanent = self._strip(main_module, keyed=False)
+        assert modified is True
+        assert permanent is True

tests/e2e/test_proxy_auth_middleware.py

@@ -12,6 +12,11 @@ import pytest
 pytestmark = pytest.mark.e2e
 
 
+def _hook(main_module, name):
+    """The registered before_request hook called `name` (it lives in core/auth_middleware.py)."""
+    return next(f for f in main_module.app.before_request_funcs[None] if f.__name__ == name)
+
+
 def _as_response(result: Any):
     if isinstance(result, tuple) and len(result) == 2:
         resp, status = result
@@ -61,7 +66,7 @@ class TestProxyAuthMiddleware:
             patch.object(main_module, "get_auth_mode", return_value="builtin"),
             main_module.app.test_request_context("/api/releases"),
         ):
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
             assert "user_id" not in main_module.session
 
@@ -70,7 +75,7 @@ class TestProxyAuthMiddleware:
             patch.object(main_module, "get_auth_mode", return_value="proxy"),
             main_module.app.test_request_context("/api/health"),
         ):
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
 
     def test_allows_auth_check_without_header(self, main_module):
@@ -83,7 +88,7 @@ class TestProxyAuthMiddleware:
             ),
             main_module.app.test_request_context("/api/auth/check"),
         ):
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
             assert "user_id" not in main_module.session
 
@@ -100,7 +105,7 @@ class TestProxyAuthMiddleware:
                 headers={"X-Auth-User": "proxyuser"},
             ),
         ):
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
             assert main_module.session.get("user_id") == "proxyuser"
             assert main_module.session.get("is_admin") is True
@@ -125,7 +130,7 @@ class TestProxyAuthMiddleware:
                 environ_base={"REMOTE_USER": "proxyremote"},
             ),
         ):
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
             assert main_module.session.get("user_id") == "proxyremote"
             assert main_module.session.get("is_admin") is True
@@ -150,7 +155,7 @@ class TestProxyAuthMiddleware:
                 headers={"X-Auth-User": "proxy_takeover_local"},
             ),
         ):
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
 
             db_user_id = main_module.session.get("db_user_id")
@@ -176,7 +181,7 @@ class TestProxyAuthMiddleware:
             main_module.session["user_id"] = "old-user"
             main_module.session["db_user_id"] = 999999
 
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
             assert main_module.session.get("user_id") == "proxyuser2"
             db_user_id = main_module.session.get("db_user_id")
@@ -203,7 +208,7 @@ class TestProxyAuthMiddleware:
             main_module.session["user_id"] = username
             main_module.session["db_user_id"] = stale_user_id
 
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
             assert main_module.session.get("user_id") == username
 
@@ -238,7 +243,7 @@ class TestProxyAuthMiddleware:
             main_module.session["user_id"] = username
             main_module.session["db_user_id"] = other_user["id"]
 
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
             assert main_module.session.get("user_id") == username
 
@@ -264,7 +269,7 @@ class TestProxyAuthMiddleware:
                 "/api/releases",
                 headers={"X-Auth-User": "first_proxy_user"},
             ):
-                assert main_module.proxy_auth_middleware() is None
+                assert _hook(main_module, "proxy_auth_middleware")() is None
                 assert main_module.session.get("is_admin") is True
 
             # The instance now has an admin, so nobody is at risk of being locked
@@ -273,7 +278,7 @@ class TestProxyAuthMiddleware:
                 "/api/releases",
                 headers={"X-Auth-User": "second_proxy_user"},
             ):
-                assert main_module.proxy_auth_middleware() is None
+                assert _hook(main_module, "proxy_auth_middleware")() is None
                 assert main_module.session.get("is_admin") is False
 
     def test_default_role_admin_promotes_later_accounts(self, main_module):
@@ -300,7 +305,7 @@ class TestProxyAuthMiddleware:
                 headers={"X-Auth-User": "later_proxy_admin"},
             ),
         ):
-            assert main_module.proxy_auth_middleware() is None
+            assert _hook(main_module, "proxy_auth_middleware")() is None
             assert main_module.session.get("is_admin") is True
 
     def test_existing_user_keeps_its_stored_role(self, main_module):
@@ -332,7 +337,7 @@ class TestProxyAuthMiddleware:
                 headers={"X-Auth-User": "known_plain_user"},
             ),
         ):
-            assert main_module.proxy_auth_middleware() is None
+            assert _hook(main_module, "proxy_auth_middleware")() is None
             # The stored role wins; the default only applies to new accounts.
             assert main_module.session.get("is_admin") is False
 
@@ -346,7 +351,7 @@ class TestProxyAuthMiddleware:
             ),
             main_module.app.test_request_context("/api/releases"),
         ):
-            resp = _as_response(main_module.proxy_auth_middleware())
+            resp = _as_response(_hook(main_module, "proxy_auth_middleware")())
             data = resp.get_json()
 
         assert resp.status_code == 401
@@ -374,7 +379,7 @@ class TestProxyAuthMiddleware:
                 },
             ),
         ):
-            result = main_module.proxy_auth_middleware()
+            result = _hook(main_module, "proxy_auth_middleware")()
             assert result is None
             assert main_module.session.get("is_admin") is True