diff --git a/shelfmark/core/auth_routes.py b/shelfmark/core/auth_routes.py new file mode 100644 index 00000000..c313a825 --- /dev/null +++ b/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, + } + ) diff --git a/shelfmark/main.py b/shelfmark/main.py index 583f0539..50cb09be 100644 --- a/shelfmark/main.py +++ b/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.`. +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: diff --git a/tests/core/test_auth_routes.py b/tests/core/test_auth_routes.py new file mode 100644 index 00000000..27612a6d --- /dev/null +++ b/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"]]