From 2254e2bf9325b9a6fc87152cd9f7ff0b9c998dd2 Mon Sep 17 00:00:00 2001 From: splitsec2 <35583321+splitsec2@users.noreply.github.com> Date: Sun, 1 Nov 2026 19:50:36 -0600 Subject: [PATCH] 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. ## 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. --- shelfmark/core/auth_middleware.py | 278 ++++++++++++++++++++ shelfmark/main.py | 237 +---------------- tests/core/test_api_key_env.py | 9 +- tests/core/test_auth_api.py | 7 +- tests/core/test_auth_middleware.py | 324 ++++++++++++++++++++++++ tests/e2e/test_proxy_auth_middleware.py | 35 +-- 6 files changed, 643 insertions(+), 247 deletions(-) create mode 100644 shelfmark/core/auth_middleware.py create mode 100644 tests/core/test_auth_middleware.py diff --git a/shelfmark/core/auth_middleware.py b/shelfmark/core/auth_middleware.py new file mode 100644 index 00000000..119e4f8e --- /dev/null +++ b/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 diff --git a/shelfmark/main.py b/shelfmark/main.py index f0bd5a64..1f9aace6 100644 --- a/shelfmark/main.py +++ b/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( diff --git a/tests/core/test_api_key_env.py b/tests/core/test_api_key_env.py index cf19cbf3..efd52b1a 100644 --- a/tests/core/test_api_key_env.py +++ b/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 diff --git a/tests/core/test_auth_api.py b/tests/core/test_auth_api.py index 4e568cc0..693e0a0e 100644 --- a/tests/core/test_auth_api.py +++ b/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 diff --git a/tests/core/test_auth_middleware.py b/tests/core/test_auth_middleware.py new file mode 100644 index 00000000..a56d2313 --- /dev/null +++ b/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 diff --git a/tests/e2e/test_proxy_auth_middleware.py b/tests/e2e/test_proxy_auth_middleware.py index 23047455..adc26111 100644 --- a/tests/e2e/test_proxy_auth_middleware.py +++ b/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