From 0d7a12ca7cb788e0e6193ba0bb5e06f6f72fcc88 Mon Sep 17 00:00:00 2001 From: Tag Howard Date: Thu, 15 Jan 2026 08:27:50 -0500 Subject: [PATCH] Feature: Reverse proxy authentication (#455) - Changes the auth settings to support more than two auth types - Added a proxy auth type with settings for user and optionally group headers - Added a global middleware `proxy_auth_middleware` to handle proxy auth (it does nothing if any other auth mode is set) - Added support for proxy auth to `get_auth_mode`, `login_required`, `api_login/out`, and `api_auth_check` - Added a backend check to make protect the API for settings when admin is required --------- Co-authored-by: Joshua Tag Howard Co-authored-by: Alex --- readme.md | 12 +- shelfmark/config/security.py | 150 +++++++++- shelfmark/main.py | 135 ++++++++- src/frontend/src/hooks/useAuth.ts | 6 +- src/frontend/src/types/index.ts | 1 + tests/config/test_security.py | 366 ++++++++++++++++++++++++ tests/e2e/test_auth_endpoints.py | 253 ++++++++++++++++ tests/e2e/test_auth_flow.py | 277 ++++++++++++++++++ tests/e2e/test_proxy_auth_middleware.py | 183 ++++++++++++ 9 files changed, 1355 insertions(+), 28 deletions(-) create mode 100644 tests/config/test_security.py create mode 100644 tests/e2e/test_auth_endpoints.py create mode 100644 tests/e2e/test_auth_flow.py create mode 100644 tests/e2e/test_proxy_auth_middleware.py diff --git a/readme.md b/readme.md index 60d300f7..afaaff0b 100644 --- a/readme.md +++ b/readme.md @@ -146,9 +146,17 @@ If you need Cloudflare bypass with the Lite image, configure an external resolve ## 🔐 Authentication -Authentication is optional but recommended for shared or exposed instances. Enable in Settings. +Authentication is optional but recommended for shared or exposed instances. Three authentication methods are available in Settings: -**Alternative**: If you're running Calibre-Web, you can reuse its user database by mounting it: +**1. Single Username/Password** + +**2. Proxy (Forward) Authentication** + +Proxy auth trusts headers set by your reverse proxy (e.g. `X-Auth-User`). Ensure Shelfmark is not directly exposed, and configure your proxy to strip/overwrite these headers for all inbound requests. + +**3. Calibre-Web Database** + +If you're running Calibre-Web, you can reuse its user database by mounting it: ```yaml volumes: diff --git a/shelfmark/config/security.py b/shelfmark/config/security.py index 5d5bc3a0..7b357359 100644 --- a/shelfmark/config/security.py +++ b/shelfmark/config/security.py @@ -10,6 +10,7 @@ from shelfmark.core.settings_registry import ( register_on_save, load_config_file, TextField, + SelectField, PasswordField, CheckboxField, ActionButton, @@ -18,6 +19,65 @@ from shelfmark.core.settings_registry import ( logger = setup_logger(__name__) +def _migrate_security_settings() -> None: + import json + from shelfmark.core.settings_registry import _get_config_file_path, _ensure_config_dir + + try: + config = load_config_file("security") + migrated = False + + # Migrate USE_CWA_AUTH to AUTH_METHOD + if "USE_CWA_AUTH" in config: + old_value = config.pop("USE_CWA_AUTH") + + # Only set AUTH_METHOD if it doesn't already exist + if "AUTH_METHOD" not in config: + if old_value: + config["AUTH_METHOD"] = "cwa" + logger.info("Migrated USE_CWA_AUTH=True to AUTH_METHOD='cwa'") + else: + # If USE_CWA_AUTH was False, determine auth method from credentials + if config.get("BUILTIN_USERNAME") and config.get("BUILTIN_PASSWORD_HASH"): + config["AUTH_METHOD"] = "builtin" + logger.info("Migrated USE_CWA_AUTH=False to AUTH_METHOD='builtin'") + else: + config["AUTH_METHOD"] = "none" + logger.info("Migrated USE_CWA_AUTH=False to AUTH_METHOD='none'") + migrated = True + else: + logger.info("Removed deprecated USE_CWA_AUTH setting (AUTH_METHOD already exists)") + migrated = True + + # Migrate RESTRICT_SETTINGS_TO_ADMIN to CWA_RESTRICT_SETTINGS_TO_ADMIN + if "RESTRICT_SETTINGS_TO_ADMIN" in config: + old_value = config.pop("RESTRICT_SETTINGS_TO_ADMIN") + + # Only migrate if new key doesn't exist + if "CWA_RESTRICT_SETTINGS_TO_ADMIN" not in config: + config["CWA_RESTRICT_SETTINGS_TO_ADMIN"] = old_value + logger.info(f"Migrated RESTRICT_SETTINGS_TO_ADMIN={old_value} to CWA_RESTRICT_SETTINGS_TO_ADMIN={old_value}") + migrated = True + else: + logger.info("Removed deprecated RESTRICT_SETTINGS_TO_ADMIN setting (CWA_RESTRICT_SETTINGS_TO_ADMIN already exists)") + migrated = True + + # Save config if any migrations occurred + if migrated: + _ensure_config_dir("security") + config_path = _get_config_file_path("security") + with open(config_path, 'w') as f: + json.dump(config, f, indent=2) + logger.info("Security settings migration completed successfully") + else: + logger.debug("No security settings migration needed") + + except FileNotFoundError: + logger.debug("No existing security config file found - nothing to migrate") + except Exception as e: + logger.error(f"Failed to migrate security settings: {e}") + + def _clear_builtin_credentials() -> Dict[str, Any]: """Clear built-in credentials to allow public access.""" import json @@ -98,20 +158,40 @@ def _on_save_security(values: Dict[str, Any]) -> Dict[str, Any]: @register_settings("security", "Security", icon="shield", order=5) -def security_settings(): +def security_settings(): """Security and authentication settings.""" from shelfmark.config.env import CWA_DB_PATH cwa_db_available = CWA_DB_PATH is not None and CWA_DB_PATH.exists() + auth_method_options = [ + {"label": "No Authentication", "value": "none"}, + {"label": "Username/Password", "value": "builtin"}, + {"label": "Proxy Authentication", "value": "proxy"}, + ] + if cwa_db_available: + auth_method_options.append({"label": "Calibre-Web Database", "value": "cwa"}) + + auth_method_description = "Select the authentication method for accessing Shelfmark." + if not cwa_db_available: + auth_method_description += " Calibre-Web database option requires mounting your Calibre-Web app.db to /auth/app.db." + fields = [ + SelectField( + key="AUTH_METHOD", + label="Authentication Method", + description=auth_method_description, + options=auth_method_options, + default="none", + env_supported=False, + ), TextField( key="BUILTIN_USERNAME", label="Username", description="Set a username and password to require login. Leave both empty for public access.", placeholder="Enter username", env_supported=False, - disabled_when={"field": "USE_CWA_AUTH", "value": True, "reason": "Using Calibre-Web database for authentication."}, + show_when={"field": "AUTH_METHOD", "value": "builtin"}, ), PasswordField( key="BUILTIN_PASSWORD", @@ -119,14 +199,14 @@ def security_settings(): description="Fill in to set or change the password.", placeholder="Enter new password", env_supported=False, - disabled_when={"field": "USE_CWA_AUTH", "value": True, "reason": "Using Calibre-Web database for authentication."}, + show_when={"field": "AUTH_METHOD", "value": "builtin"}, ), PasswordField( key="BUILTIN_PASSWORD_CONFIRM", label="Confirm Password", placeholder="Confirm new password", env_supported=False, - disabled_when={"field": "USE_CWA_AUTH", "value": True, "reason": "Using Calibre-Web database for authentication."}, + show_when={"field": "AUTH_METHOD", "value": "builtin"}, ), ActionButton( key="clear_credentials", @@ -134,28 +214,72 @@ def security_settings(): description="Remove login requirement and make the app publicly accessible.", style="danger", callback=_clear_builtin_credentials, - disabled_when={"field": "USE_CWA_AUTH", "value": True, "reason": "Using Calibre-Web database for authentication."}, + show_when={"field": "AUTH_METHOD", "value": "builtin"}, + ), + TextField( + key="PROXY_AUTH_USER_HEADER", + label="Proxy Auth User Header", + description=( + "The HTTP header your proxy uses to pass the authenticated username." + ), + placeholder="e.g. X-Auth-User", + default="X-Auth-User", + env_supported=False, + show_when={"field": "AUTH_METHOD", "value": "proxy"}, + ), + TextField( + key="PROXY_AUTH_LOGOUT_URL", + label="Proxy Auth Logout URL", + description=( + "The URL to redirect users to for logging out." + " Leave empty to disable logout functionality." + ), + placeholder="https://myauth.example.com/logout", + default="", + env_supported=False, + show_when={"field": "AUTH_METHOD", "value": "proxy"}, ), CheckboxField( - key="USE_CWA_AUTH", - label="Use Calibre-Web Database", + key="PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN", + label="Restrict Settings to Admins authenticated via Proxy", description=( - "Use your existing Calibre-Web user credentials for authentication." + "Only users in the admin group can access settings." ), default=False, env_supported=False, - disabled=not cwa_db_available, - disabled_reason="Mount your Calibre-Web app.db to /auth/app.db in docker compose to enable.", + show_when={"field": "AUTH_METHOD", "value": "proxy"}, + ), + TextField( + key="PROXY_AUTH_ADMIN_GROUP_HEADER", + label="Proxy Auth Admin Group Header", + description=( + "The HTTP header your proxy uses to pass the user's groups/roles." + ), + placeholder="e.g. X-Auth-Groups", + default="X-Auth-Groups", + env_supported=False, + show_when={"field": "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN", "value": True}, + ), + TextField( + key="PROXY_AUTH_ADMIN_GROUP_NAME", + label="Proxy Auth Admin Group Name", + description=( + "The name of the group/role that should have admin access." + ), + placeholder="e.g. admins", + default="admins", + env_supported=False, + show_when={"field": "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN", "value": True}, ), CheckboxField( - key="RESTRICT_SETTINGS_TO_ADMIN", - label="Restrict Settings to Admins", + key="CWA_RESTRICT_SETTINGS_TO_ADMIN", + label="Restrict Settings to Admins authenticated via Calibre-Web", description=( "Only users with admin role in Calibre-Web can access settings." ), default=False, env_supported=False, - show_when={"field": "USE_CWA_AUTH", "value": True}, + show_when={"field": "AUTH_METHOD", "value": "cwa"}, ), ] diff --git a/shelfmark/main.py b/shelfmark/main.py index 626935ff..b34bf264 100644 --- a/shelfmark/main.py +++ b/shelfmark/main.py @@ -77,6 +77,10 @@ try: except ImportError as e: logger.warning(f"Failed to import plugin modules: {e}") +# Migrate legacy security settings if needed +from shelfmark.config.security import _migrate_security_settings +_migrate_security_settings() + # Start download coordinator backend.start() @@ -156,12 +160,13 @@ def get_auth_mode() -> str: try: security_config = load_config_file("security") - # 1. Check for explicit CWA auth (CWA_DB_PATH is pre-validated at startup) - if security_config.get("USE_CWA_AUTH") and CWA_DB_PATH: + auth_mode = security_config.get("AUTH_METHOD", "none") + if auth_mode == "cwa" and CWA_DB_PATH: return "cwa" - # 2. Check for built-in credentials - if security_config.get("BUILTIN_USERNAME") and security_config.get("BUILTIN_PASSWORD_HASH"): + if auth_mode == "builtin" and security_config.get("BUILTIN_USERNAME") and security_config.get("BUILTIN_PASSWORD_HASH"): return "builtin" + if auth_mode == "proxy" and security_config.get("PROXY_AUTH_USER_HEADER"): + return "proxy" except Exception: pass @@ -237,6 +242,66 @@ app.config.update( logger.info(f"Session cookie secure setting: {SESSION_COOKIE_SECURE} (from env: {SESSION_COOKIE_SECURE_ENV})") +@app.before_request +def proxy_auth_middleware(): + """ + 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 + + # Skip for public endpoints that don't need auth + if request.path == '/api/health': + return None + + from shelfmark.core.settings_registry import load_config_file + + try: + security_config = load_config_file("security") + user_header = security_config.get("PROXY_AUTH_USER_HEADER", "X-Auth-User") + + # Extract username from proxy header + username = request.headers.get(user_header) + + if not username: + if request.path.startswith('/api/auth/'): + return None + + logger.warning(f"Proxy auth enabled but no username found in header '{user_header}'") + return jsonify({"error": "Authentication required. Proxy header not set."}), 401 + + # Check if settings access should be restricted to admins + restrict_to_admin = security_config.get("PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN", False) + is_admin = True # Default to admin if not restricting + + if restrict_to_admin: + admin_group_header = security_config.get("PROXY_AUTH_ADMIN_GROUP_HEADER", "X-Auth-Groups") + admin_group_name = security_config.get("PROXY_AUTH_ADMIN_GROUP_NAME", "admins") + + # Extract groups from proxy header (can be comma or pipe separated) + groups_header = request.headers.get(admin_group_header, "") + 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 + + # Create or update session + session['user_id'] = username + session['is_admin'] = is_admin + session.permanent = False + + return None + + except Exception as e: + logger.error(f"Proxy auth middleware error: {e}") + return jsonify({"error": "Authentication error"}), 500 + def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): @@ -255,6 +320,25 @@ def login_required(f): if 'user_id' not in session: return jsonify({"error": "Unauthorized"}), 401 + # Check admin access for settings endpoints (proxy and CWA modes) + if auth_mode in ("proxy", "cwa") and (request.path.startswith('/api/settings') or request.path.startswith('/api/onboarding')): + from shelfmark.core.settings_registry import load_config_file + + try: + security_config = load_config_file("security") + + if auth_mode == "proxy": + restrict_to_admin = security_config.get("PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN", False) + else: + restrict_to_admin = security_config.get("CWA_RESTRICT_SETTINGS_TO_ADMIN", False) + + if restrict_to_admin and not session.get('is_admin', False): + return jsonify({"error": "Admin access required"}), 403 + + except Exception as e: + logger.error(f"Admin access check error: {e}") + return jsonify({"error": "Internal Server Error"}), 500 + return f(*args, **kwargs) return decorated_function @@ -870,6 +954,10 @@ def api_login() -> Union[Response, Tuple[Response, int]]: 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 + username = data.get('username', '').strip() password = data.get('password', '') remember_me = data.get('remember_me', False) @@ -886,8 +974,6 @@ def api_login() -> Union[Response, Tuple[Response, int]]: "error": f"Account temporarily locked due to multiple failed login attempts. Try again in {int(remaining_time)} minutes." }), 429 - auth_mode = get_auth_mode() - # If no authentication is configured, authentication always succeeds if auth_mode == "none": session['user_id'] = username @@ -964,15 +1050,27 @@ def api_login() -> Union[Response, Tuple[Response, int]]: def api_logout() -> Union[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. + flask.Response: JSON with success status and optional logout_url. """ + from shelfmark.core.settings_registry import load_config_file + try: + auth_mode = get_auth_mode() ip_address = get_client_ip() username = session.get('user_id', 'unknown') session.clear() logger.info(f"Logout successful for user '{username}' from IP {ip_address}") + + # For proxy auth, include logout URL if configured + if auth_mode == "proxy": + security_config = load_config_file("security") + logout_url = security_config.get("PROXY_AUTH_LOGOUT_URL", "") + if logout_url: + return jsonify({"success": True, "logout_url": logout_url}) + return jsonify({"success": True}) except Exception as e: logger.error_trace(f"Logout error: {e}") @@ -990,6 +1088,7 @@ def api_auth_check() -> Union[Response, Tuple[Response, int]]: from shelfmark.core.settings_registry import load_config_file try: + security_config = load_config_file("security") auth_mode = get_auth_mode() # If no authentication is configured, access is allowed (full admin) @@ -1007,25 +1106,37 @@ def api_auth_check() -> Union[Response, Tuple[Response, int]]: # Determine admin status for settings access # - Built-in auth: single user is always admin # - CWA auth: check RESTRICT_SETTINGS_TO_ADMIN setting + # - Proxy auth: check PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN setting if auth_mode == "builtin": is_admin = True elif auth_mode == "cwa": - security_config = load_config_file("security") - restrict_to_admin = security_config.get("RESTRICT_SETTINGS_TO_ADMIN", False) + restrict_to_admin = security_config.get("CWA_RESTRICT_SETTINGS_TO_ADMIN", False) if restrict_to_admin: is_admin = session.get('is_admin', False) else: # All authenticated CWA users can access settings is_admin = True + elif auth_mode == "proxy": + restrict_to_admin = security_config.get("PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN", False) + is_admin = session.get('is_admin', not restrict_to_admin) else: is_admin = False - return jsonify({ + response_data = { "authenticated": is_authenticated, "auth_required": True, "auth_mode": auth_mode, - "is_admin": is_admin if is_authenticated else False - }) + "is_admin": is_admin if is_authenticated else False, + "username": session.get('user_id') if is_authenticated else None + } + + # Add logout URL for proxy auth if configured + if auth_mode == "proxy" and security_config.get("PROXY_AUTH_USER_HEADER"): + logout_url = security_config.get("PROXY_AUTH_LOGOUT_URL", "") + if logout_url: + response_data["logout_url"] = logout_url + + return jsonify(response_data) except Exception as e: logger.error_trace(f"Auth check error: {e}") return jsonify({ diff --git a/src/frontend/src/hooks/useAuth.ts b/src/frontend/src/hooks/useAuth.ts index 08dfddfb..54f1a194 100644 --- a/src/frontend/src/hooks/useAuth.ts +++ b/src/frontend/src/hooks/useAuth.ts @@ -83,7 +83,11 @@ export function useAuth(options: UseAuthOptions = {}): UseAuthReturn { const handleLogout = useCallback(async () => { try { - await logout(); + const { logout_url } = await logout(); + if (logout_url?.startsWith('https://') || logout_url?.startsWith('http://')) { + window.location.href = logout_url; + return; + } setIsAuthenticated(false); onLogoutSuccess?.(); navigate('/login', { replace: true }); diff --git a/src/frontend/src/types/index.ts b/src/frontend/src/types/index.ts index 0e1527a2..727aab12 100644 --- a/src/frontend/src/types/index.ts +++ b/src/frontend/src/types/index.ts @@ -181,6 +181,7 @@ export interface AuthResponse { auth_required?: boolean; is_admin?: boolean; error?: string; + logout_url?: string; } // Type guard to check if a book is from a metadata provider diff --git a/tests/config/test_security.py b/tests/config/test_security.py new file mode 100644 index 00000000..392db1fb --- /dev/null +++ b/tests/config/test_security.py @@ -0,0 +1,366 @@ +""" +Tests for security configuration and migration. + +Tests the security settings registration, migration from old settings, +and proxy authentication configuration. +""" + +import json +import os +import tempfile +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture +def temp_config_dir(): + """Create a temporary config directory for tests.""" + with tempfile.TemporaryDirectory() as tmpdir: + config_dir = Path(tmpdir) + security_dir = config_dir / "security" + security_dir.mkdir(parents=True, exist_ok=True) + yield security_dir + + +@pytest.fixture +def mock_logger(): + """Mock logger to capture log messages.""" + return MagicMock() + + +class TestSecurityMigration: + """Tests for migrating legacy security settings.""" + + def test_migrate_use_cwa_auth_true(self, temp_config_dir, mock_logger): + """Test migrating USE_CWA_AUTH=True to AUTH_METHOD='cwa'.""" + # Create legacy config + config_file = temp_config_dir / "config.json" + legacy_config = { + "USE_CWA_AUTH": True, + "BUILTIN_USERNAME": "admin", + "BUILTIN_PASSWORD_HASH": "hashed_password" + } + config_file.write_text(json.dumps(legacy_config, indent=2)) + + # Mock load_config_file to return our test config, and the paths + with patch('shelfmark.config.security.load_config_file', return_value=legacy_config.copy()): + with patch('shelfmark.core.settings_registry._get_config_file_path', return_value=str(config_file)): + with patch('shelfmark.core.settings_registry._ensure_config_dir'): + with patch('shelfmark.config.security.logger', mock_logger): + from shelfmark.config.security import _migrate_security_settings + _migrate_security_settings() + + # Verify migration - read the actual file + migrated_config = json.loads(config_file.read_text()) + assert migrated_config["AUTH_METHOD"] == "cwa" + assert "USE_CWA_AUTH" not in migrated_config + + def test_migrate_use_cwa_auth_false_with_credentials(self, temp_config_dir, mock_logger): + """Test migrating USE_CWA_AUTH=False with credentials to AUTH_METHOD='builtin'.""" + config_file = temp_config_dir / "config.json" + legacy_config = { + "USE_CWA_AUTH": False, + "BUILTIN_USERNAME": "admin", + "BUILTIN_PASSWORD_HASH": "hashed_password" + } + config_file.write_text(json.dumps(legacy_config, indent=2)) + + with patch('shelfmark.config.security.load_config_file', return_value=legacy_config.copy()): + with patch('shelfmark.core.settings_registry._get_config_file_path', return_value=str(config_file)): + with patch('shelfmark.core.settings_registry._ensure_config_dir'): + with patch('shelfmark.config.security.logger', mock_logger): + from shelfmark.config.security import _migrate_security_settings + _migrate_security_settings() + + migrated_config = json.loads(config_file.read_text()) + assert migrated_config["AUTH_METHOD"] == "builtin" + assert "USE_CWA_AUTH" not in migrated_config + + def test_migrate_use_cwa_auth_false_without_credentials(self, temp_config_dir, mock_logger): + """Test migrating USE_CWA_AUTH=False without credentials to AUTH_METHOD='none'.""" + config_file = temp_config_dir / "config.json" + legacy_config = { + "USE_CWA_AUTH": False + } + config_file.write_text(json.dumps(legacy_config, indent=2)) + + with patch('shelfmark.config.security.load_config_file', return_value=legacy_config.copy()): + with patch('shelfmark.core.settings_registry._get_config_file_path', return_value=str(config_file)): + with patch('shelfmark.core.settings_registry._ensure_config_dir'): + with patch('shelfmark.config.security.logger', mock_logger): + from shelfmark.config.security import _migrate_security_settings + _migrate_security_settings() + + migrated_config = json.loads(config_file.read_text()) + assert migrated_config["AUTH_METHOD"] == "none" + assert "USE_CWA_AUTH" not in migrated_config + + def test_migrate_restrict_settings_to_admin(self, temp_config_dir, mock_logger): + """Test migrating RESTRICT_SETTINGS_TO_ADMIN to CWA_RESTRICT_SETTINGS_TO_ADMIN.""" + config_file = temp_config_dir / "config.json" + legacy_config = { + "AUTH_METHOD": "cwa", + "RESTRICT_SETTINGS_TO_ADMIN": True + } + config_file.write_text(json.dumps(legacy_config, indent=2)) + + with patch('shelfmark.config.security.load_config_file', return_value=legacy_config.copy()): + with patch('shelfmark.core.settings_registry._get_config_file_path', return_value=str(config_file)): + with patch('shelfmark.core.settings_registry._ensure_config_dir'): + with patch('shelfmark.config.security.logger', mock_logger): + from shelfmark.config.security import _migrate_security_settings + _migrate_security_settings() + + migrated_config = json.loads(config_file.read_text()) + assert migrated_config["CWA_RESTRICT_SETTINGS_TO_ADMIN"] is True + assert "RESTRICT_SETTINGS_TO_ADMIN" not in migrated_config + + def test_migrate_preserves_existing_auth_method(self, temp_config_dir, mock_logger): + """Test that existing AUTH_METHOD is not overwritten during migration.""" + config_file = temp_config_dir / "config.json" + legacy_config = { + "USE_CWA_AUTH": True, + "AUTH_METHOD": "proxy" # Already has new format + } + config_file.write_text(json.dumps(legacy_config, indent=2)) + + with patch('shelfmark.config.security.load_config_file', return_value=legacy_config.copy()): + with patch('shelfmark.core.settings_registry._get_config_file_path', return_value=str(config_file)): + with patch('shelfmark.core.settings_registry._ensure_config_dir'): + with patch('shelfmark.config.security.logger', mock_logger): + from shelfmark.config.security import _migrate_security_settings + _migrate_security_settings() + + migrated_config = json.loads(config_file.read_text()) + assert migrated_config["AUTH_METHOD"] == "proxy" # Should not change + assert "USE_CWA_AUTH" not in migrated_config + + def test_migrate_handles_missing_config_file(self, temp_config_dir, mock_logger): + """Test that migration handles missing config file gracefully.""" + with patch('shelfmark.config.security.load_config_file', side_effect=FileNotFoundError()): + with patch('shelfmark.config.security.logger', mock_logger): + from shelfmark.config.security import _migrate_security_settings + _migrate_security_settings() + + mock_logger.debug.assert_any_call("No existing security config file found - nothing to migrate") + + def test_migrate_no_changes_needed(self, temp_config_dir, mock_logger): + """Test migration when no changes are needed.""" + config_file = temp_config_dir / "config.json" + modern_config = { + "AUTH_METHOD": "builtin", + "BUILTIN_USERNAME": "admin", + "BUILTIN_PASSWORD_HASH": "hashed_password" + } + config_file.write_text(json.dumps(modern_config, indent=2)) + + with patch('shelfmark.config.security.load_config_file', return_value=modern_config.copy()): + with patch('shelfmark.core.settings_registry._get_config_file_path', return_value=str(config_file)): + with patch('shelfmark.core.settings_registry._ensure_config_dir'): + with patch('shelfmark.config.security.logger', mock_logger): + from shelfmark.config.security import _migrate_security_settings + _migrate_security_settings() + + # Config should remain unchanged + final_config = json.loads(config_file.read_text()) + # File won't have been rewritten, so it should be the original + assert final_config == modern_config + mock_logger.debug.assert_any_call("No security settings migration needed") + + +class TestSecuritySettings: + """Tests for security settings registration.""" + + def test_security_settings_without_cwa(self): + """Test that CWA option is not available when DB is not mounted.""" + # Patch CWA_DB_PATH where it's imported in the function + with patch('shelfmark.config.env.CWA_DB_PATH', None): + # Need to reload the module to pick up the patch + import importlib + import shelfmark.config.security + importlib.reload(shelfmark.config.security) + from shelfmark.config.security import security_settings + + fields = security_settings() + + # Find the AUTH_METHOD field + auth_method_field = next((f for f in fields if f.key == "AUTH_METHOD"), None) + assert auth_method_field is not None + + # CWA should not be in options + option_values = [opt["value"] for opt in auth_method_field.options] + assert "none" in option_values + assert "builtin" in option_values + assert "proxy" in option_values + assert "cwa" not in option_values + + def test_security_settings_with_cwa(self): + """Test that CWA option is available when DB is mounted.""" + # Create a mock path that exists + mock_path = MagicMock() + mock_path.exists.return_value = True + + with patch('shelfmark.config.env.CWA_DB_PATH', mock_path): + import importlib + import shelfmark.config.security + importlib.reload(shelfmark.config.security) + from shelfmark.config.security import security_settings + + fields = security_settings() + + # Find the AUTH_METHOD field + auth_method_field = next((f for f in fields if f.key == "AUTH_METHOD"), None) + assert auth_method_field is not None + + # CWA should be in options + option_values = [opt["value"] for opt in auth_method_field.options] + assert "cwa" in option_values + + def test_proxy_auth_fields_present(self): + """Test that proxy auth configuration fields are present.""" + from shelfmark.config.security import security_settings + + fields = security_settings() + field_keys = [f.key for f in fields] + + # Verify proxy auth fields exist + assert "PROXY_AUTH_USER_HEADER" in field_keys + assert "PROXY_AUTH_LOGOUT_URL" in field_keys + assert "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN" in field_keys + assert "PROXY_AUTH_ADMIN_GROUP_HEADER" in field_keys + assert "PROXY_AUTH_ADMIN_GROUP_NAME" in field_keys + + def test_cwa_restrict_settings_field_present(self): + """Test that CWA restrict settings field is present.""" + from shelfmark.config.security import security_settings + + fields = security_settings() + field_keys = [f.key for f in fields] + + assert "CWA_RESTRICT_SETTINGS_TO_ADMIN" in field_keys + + +class TestPasswordValidation: + """Tests for password validation in the on_save handler.""" + + def test_on_save_validates_password_match(self): + """Test that passwords must match.""" + from shelfmark.config.security import _on_save_security + + values = { + "AUTH_METHOD": "builtin", + "BUILTIN_USERNAME": "admin", + "BUILTIN_PASSWORD": "password123", + "BUILTIN_PASSWORD_CONFIRM": "different_password" + } + + result = _on_save_security(values) + + assert result["error"] is True + assert "do not match" in result["message"] + + def test_on_save_validates_password_length(self): + """Test that password must be at least 4 characters.""" + from shelfmark.config.security import _on_save_security + + values = { + "AUTH_METHOD": "builtin", + "BUILTIN_USERNAME": "admin", + "BUILTIN_PASSWORD": "abc", + "BUILTIN_PASSWORD_CONFIRM": "abc" + } + + result = _on_save_security(values) + + assert result["error"] is True + assert "at least 4 characters" in result["message"] + + def test_on_save_requires_username_with_password(self): + """Test that username is required when password is set.""" + from shelfmark.config.security import _on_save_security + + values = { + "AUTH_METHOD": "builtin", + "BUILTIN_PASSWORD": "password123", + "BUILTIN_PASSWORD_CONFIRM": "password123" + } + + result = _on_save_security(values) + + assert result["error"] is True + assert "Username cannot be empty" in result["message"] + + def test_on_save_hashes_password(self): + """Test that password is properly hashed.""" + from shelfmark.config.security import _on_save_security + + values = { + "AUTH_METHOD": "builtin", + "BUILTIN_USERNAME": "admin", + "BUILTIN_PASSWORD": "password123", + "BUILTIN_PASSWORD_CONFIRM": "password123" + } + + result = _on_save_security(values) + + assert result["error"] is False + assert "BUILTIN_PASSWORD_HASH" in result["values"] + assert "BUILTIN_PASSWORD" not in result["values"] + assert "BUILTIN_PASSWORD_CONFIRM" not in result["values"] + # Hash should be different from raw password + assert result["values"]["BUILTIN_PASSWORD_HASH"] != "password123" + + def test_on_save_preserves_existing_hash_when_no_password(self): + """Test that existing password hash is preserved when password fields are empty.""" + from shelfmark.config.security import _on_save_security + + with patch('shelfmark.config.security.load_config_file') as mock_load: + mock_load.return_value = { + "BUILTIN_PASSWORD_HASH": "existing_hash" + } + + values = { + "AUTH_METHOD": "builtin", + "BUILTIN_USERNAME": "admin" + } + + result = _on_save_security(values) + + assert result["error"] is False + assert result["values"]["BUILTIN_PASSWORD_HASH"] == "existing_hash" + + +class TestClearCredentials: + """Tests for clearing built-in credentials.""" + + def test_clear_credentials_removes_username_and_hash(self, temp_config_dir): + """Test that clearing credentials removes username and password hash.""" + config_file = temp_config_dir / "config.json" + config = { + "AUTH_METHOD": "builtin", + "BUILTIN_USERNAME": "admin", + "BUILTIN_PASSWORD_HASH": "hashed_password" + } + config_file.write_text(json.dumps(config, indent=2)) + + with patch('shelfmark.core.settings_registry._get_config_file_path', return_value=str(config_file)): + with patch('shelfmark.core.settings_registry._ensure_config_dir'): + with patch('shelfmark.config.security.load_config_file', return_value=config.copy()): + from shelfmark.config.security import _clear_builtin_credentials + result = _clear_builtin_credentials() + + assert result["success"] is True + cleared_config = json.loads(config_file.read_text()) + assert "BUILTIN_USERNAME" not in cleared_config + assert "BUILTIN_PASSWORD_HASH" not in cleared_config + + def test_clear_credentials_handles_errors(self): + """Test that clearing credentials handles errors gracefully.""" + with patch('shelfmark.config.security.load_config_file', side_effect=Exception("Test error")): + from shelfmark.config.security import _clear_builtin_credentials + result = _clear_builtin_credentials() + + assert result["success"] is False + assert "Test error" in result["message"] diff --git a/tests/e2e/test_auth_endpoints.py b/tests/e2e/test_auth_endpoints.py new file mode 100644 index 00000000..96cd84fc --- /dev/null +++ b/tests/e2e/test_auth_endpoints.py @@ -0,0 +1,253 @@ +"""Unit tests for authentication endpoints. + +These tests exercise the Flask route functions in `shelfmark.main` using Flask +request contexts. They do not require the full application stack. +""" + +from __future__ import annotations + +import importlib +from datetime import datetime, timedelta +from typing import Any, Tuple +from unittest.mock import Mock, patch + +import pytest + + +def _as_response(result: Any): + """Normalize Flask view return values to a Response-like object.""" + if isinstance(result, tuple) and len(result) == 2: + resp, status = result + resp.status_code = status + return resp + return result + + +@pytest.fixture(scope="module") +def main_module(): + """Import `shelfmark.main` with background thread startup disabled.""" + with patch("shelfmark.download.orchestrator.start"): + import shelfmark.main as main + + # Reload to ensure patched orchestrator.start is used even if imported elsewhere. + importlib.reload(main) + return main + + +class TestGetAuthMode: + def test_get_auth_mode_none(self, main_module): + with patch("shelfmark.core.settings_registry.load_config_file", return_value={"AUTH_METHOD": "none"}): + assert main_module.get_auth_mode() == "none" + + def test_get_auth_mode_builtin(self, main_module): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={ + "AUTH_METHOD": "builtin", + "BUILTIN_USERNAME": "admin", + "BUILTIN_PASSWORD_HASH": "hashed_password", + }, + ): + assert main_module.get_auth_mode() == "builtin" + + def test_get_auth_mode_proxy(self, main_module): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={"AUTH_METHOD": "proxy", "PROXY_AUTH_USER_HEADER": "X-Auth-User"}, + ): + assert main_module.get_auth_mode() == "proxy" + + def test_get_auth_mode_cwa(self, main_module): + with patch("shelfmark.core.settings_registry.load_config_file", return_value={"AUTH_METHOD": "cwa"}): + with patch.object(main_module, "CWA_DB_PATH", object()): + assert main_module.get_auth_mode() == "cwa" + + def test_get_auth_mode_default_on_error(self, main_module): + with patch("shelfmark.core.settings_registry.load_config_file", side_effect=Exception("boom")): + assert main_module.get_auth_mode() == "none" + + +class TestAuthCheckEndpoint: + def test_auth_check_no_auth(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="none"): + with patch("shelfmark.core.settings_registry.load_config_file", return_value={}): + with main_module.app.test_request_context("/api/auth/check"): + resp = _as_response(main_module.api_auth_check()) + data = resp.get_json() + + assert resp.status_code == 200 + assert data == { + "authenticated": True, + "auth_required": False, + "auth_mode": "none", + "is_admin": True, + } + + def test_auth_check_builtin_not_authenticated(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + with patch("shelfmark.core.settings_registry.load_config_file", return_value={}): + with main_module.app.test_request_context("/api/auth/check"): + resp = _as_response(main_module.api_auth_check()) + data = resp.get_json() + + assert resp.status_code == 200 + assert data["authenticated"] is False + assert data["auth_required"] is True + assert data["auth_mode"] == "builtin" + assert data["is_admin"] is False + assert data["username"] is None + + def test_auth_check_builtin_authenticated(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + with patch("shelfmark.core.settings_registry.load_config_file", return_value={}): + with main_module.app.test_request_context("/api/auth/check"): + main_module.session["user_id"] = "admin" + resp = _as_response(main_module.api_auth_check()) + data = resp.get_json() + + assert resp.status_code == 200 + assert data["authenticated"] is True + assert data["auth_required"] is True + assert data["auth_mode"] == "builtin" + assert data["is_admin"] is True + assert data["username"] == "admin" + + def test_auth_check_proxy_includes_logout_url(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="proxy"): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={ + "PROXY_AUTH_USER_HEADER": "X-Auth-User", + "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN": True, + "PROXY_AUTH_LOGOUT_URL": "https://auth.example.com/logout", + }, + ): + with main_module.app.test_request_context("/api/auth/check"): + main_module.session["user_id"] = "proxyuser" + main_module.session["is_admin"] = True + resp = _as_response(main_module.api_auth_check()) + data = resp.get_json() + + assert resp.status_code == 200 + assert data["authenticated"] is True + assert data["auth_mode"] == "proxy" + assert data["username"] == "proxyuser" + assert data["logout_url"] == "https://auth.example.com/logout" + + +class TestLoginEndpoint: + def test_login_proxy_mode_disabled(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="proxy"): + with main_module.app.test_request_context( + "/api/auth/login", + method="POST", + json={"anything": "x"}, + ): + resp = _as_response(main_module.api_login()) + data = resp.get_json() + + assert resp.status_code == 401 + assert "Proxy authentication" in (data.get("error") or "") + + def test_login_no_auth_success(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="none"): + with patch.object(main_module, "is_account_locked", return_value=False): + with main_module.app.test_request_context( + "/api/auth/login", + method="POST", + json={"username": "anyuser", "password": "anypass", "remember_me": True}, + ): + resp = _as_response(main_module.api_login()) + data = resp.get_json() + assert main_module.session.get("user_id") == "anyuser" + assert main_module.session.permanent is True + + assert resp.status_code == 200 + assert data.get("success") is True + + def test_login_builtin_success(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + with patch.object(main_module, "is_account_locked", return_value=False): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={ + "BUILTIN_USERNAME": "admin", + "BUILTIN_PASSWORD_HASH": "hash", + }, + ): + with patch.object(main_module, "check_password_hash", return_value=True): + with main_module.app.test_request_context( + "/api/auth/login", + method="POST", + json={"username": "admin", "password": "correct", "remember_me": False}, + ): + resp = _as_response(main_module.api_login()) + data = resp.get_json() + assert main_module.session.get("user_id") == "admin" + + assert resp.status_code == 200 + assert data.get("success") is True + + +class TestLogoutEndpoint: + def test_logout_proxy_returns_logout_url(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="proxy"): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={"PROXY_AUTH_LOGOUT_URL": "https://auth.example.com/logout"}, + ): + with main_module.app.test_request_context("/api/auth/logout", method="POST"): + main_module.session["user_id"] = "proxyuser" + resp = _as_response(main_module.api_logout()) + data = resp.get_json() + + assert resp.status_code == 200 + assert data["success"] is True + assert data["logout_url"] == "https://auth.example.com/logout" + + def test_logout_basic(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + with patch("shelfmark.core.settings_registry.load_config_file", return_value={}): + with main_module.app.test_request_context("/api/auth/logout", method="POST"): + main_module.session["user_id"] = "admin" + resp = _as_response(main_module.api_logout()) + data = resp.get_json() + + assert resp.status_code == 200 + assert data["success"] is True + assert "logout_url" not in data + + +class TestRateLimiting: + def test_record_failed_login_increments_count(self, main_module): + main_module.failed_login_attempts.clear() + + is_locked = main_module.record_failed_login("testuser", "127.0.0.1") + + assert is_locked is False + assert main_module.failed_login_attempts["testuser"]["count"] == 1 + + def test_account_locked_after_max_attempts(self, main_module): + main_module.failed_login_attempts.clear() + + for _ in range(main_module.MAX_LOGIN_ATTEMPTS): + is_locked = main_module.record_failed_login("testuser", "127.0.0.1") + + assert is_locked is True + assert "lockout_until" in main_module.failed_login_attempts["testuser"] + + def test_is_account_locked(self, main_module): + main_module.failed_login_attempts.clear() + main_module.failed_login_attempts["testuser"] = { + "count": 10, + "lockout_until": datetime.now() + timedelta(hours=1), + } + + assert main_module.is_account_locked("testuser") is True + + def test_clear_failed_logins(self, main_module): + main_module.failed_login_attempts["testuser"] = {"count": 5} + + main_module.clear_failed_logins("testuser") + + assert "testuser" not in main_module.failed_login_attempts diff --git a/tests/e2e/test_auth_flow.py b/tests/e2e/test_auth_flow.py new file mode 100644 index 00000000..7f96c913 --- /dev/null +++ b/tests/e2e/test_auth_flow.py @@ -0,0 +1,277 @@ +""" +E2E tests for authentication endpoints. + +Tests the full authentication flow including login, logout, and auth check +with various authentication modes. + +Run with: docker exec test-cwabd python3 -m pytest tests/e2e/ -v -m e2e +""" + +import pytest + +from .conftest import APIClient + + +@pytest.mark.e2e +class TestAuthenticationFlow: + """Tests for the authentication endpoints in a real environment.""" + + def test_auth_check_endpoint_exists(self, api_client: APIClient): + """Test that auth check endpoint is accessible.""" + resp = api_client.get("/api/auth/check") + + assert resp.status_code == 200 + data = resp.json() + assert "authenticated" in data + assert "auth_required" in data + assert "auth_mode" in data + + def test_auth_check_returns_auth_mode(self, api_client: APIClient): + """Test that auth check returns the current auth mode.""" + resp = api_client.get("/api/auth/check") + + data = resp.json() + assert "auth_mode" in data + # Should be one of the valid auth modes + assert data["auth_mode"] in ["none", "builtin", "cwa", "proxy"] + + def test_auth_check_includes_admin_status(self, api_client: APIClient): + """Test that auth check includes admin status.""" + resp = api_client.get("/api/auth/check") + + data = resp.json() + assert "is_admin" in data + assert isinstance(data["is_admin"], bool) + + def test_logout_endpoint_exists(self, api_client: APIClient): + """Test that logout endpoint is accessible.""" + resp = api_client.post("/api/auth/logout") + + # Should return 200 whether authenticated or not + assert resp.status_code == 200 + data = resp.json() + assert "success" in data + + def test_logout_may_return_logout_url(self, api_client: APIClient): + """Test that logout may return a logout URL for proxy auth.""" + resp = api_client.post("/api/auth/logout") + + data = resp.json() + # logout_url is optional depending on auth mode + if "logout_url" in data: + assert isinstance(data["logout_url"], str) + + def test_login_endpoint_exists(self, api_client: APIClient): + """Test that login endpoint is accessible.""" + resp = api_client.post("/api/auth/login", json={ + "username": "test", + "password": "test", + "remember_me": False + }) + + # Should return some response (may be success or error depending on config) + assert resp.status_code in [200, 401, 403] + + def test_login_with_no_auth_succeeds(self, api_client: APIClient): + """Test that login succeeds when no authentication is required.""" + # First check if auth is required + auth_check = api_client.get("/api/auth/check") + auth_data = auth_check.json() + + if not auth_data.get("auth_required"): + # Try logging in + resp = api_client.post("/api/auth/login", json={ + "username": "anyuser", + "password": "anypass", + "remember_me": False + }) + + # Should succeed + assert resp.status_code == 200 + data = resp.json() + assert data.get("success") is True + + +@pytest.mark.e2e +class TestProxyAuthentication: + """Tests for proxy authentication mode.""" + + def test_proxy_auth_with_valid_header(self, api_client: APIClient): + """Test proxy auth when valid user header is present.""" + # Check current auth mode + auth_check = api_client.get("/api/auth/check") + auth_data = auth_check.json() + + if auth_data.get("auth_mode") != "proxy": + pytest.skip("Proxy authentication not configured") + + # Make a request with proxy auth header + # Note: In real deployment, these headers would be set by the proxy + resp = api_client.get("/api/config", headers={"X-Auth-User": "proxyuser"}) + + if resp.status_code == 401: + pytest.skip("Proxy auth header not accepted (check proxy configuration)") + + # Should be able to access the endpoint + assert resp.status_code == 200 + + def test_proxy_auth_logout_url_available(self, api_client: APIClient): + """Test that proxy auth provides logout URL if configured.""" + # Check current auth mode + auth_check = api_client.get("/api/auth/check") + auth_data = auth_check.json() + + if auth_data.get("auth_mode") != "proxy": + pytest.skip("Proxy authentication not configured") + + # Check for logout URL in auth check response + if "logout_url" in auth_data: + assert isinstance(auth_data["logout_url"], str) + assert len(auth_data["logout_url"]) > 0 + + +@pytest.mark.e2e +class TestBuiltinAuthentication: + """Tests for built-in username/password authentication.""" + + def test_builtin_auth_requires_credentials(self, api_client: APIClient): + """Test that endpoints require authentication when builtin auth is enabled.""" + # Check current auth mode + auth_check = api_client.get("/api/auth/check") + auth_data = auth_check.json() + + if auth_data.get("auth_mode") != "builtin": + pytest.skip("Built-in authentication not configured") + + if not auth_data.get("authenticated"): + # Attempt to access protected endpoint without authentication + resp = api_client.get("/api/config") + + # Should be blocked + assert resp.status_code == 401 + + def test_builtin_auth_invalid_credentials(self, api_client: APIClient): + """Test login with invalid credentials fails.""" + # Check current auth mode + auth_check = api_client.get("/api/auth/check") + auth_data = auth_check.json() + + if auth_data.get("auth_mode") != "builtin": + pytest.skip("Built-in authentication not configured") + + # Try logging in with invalid credentials + resp = api_client.post("/api/auth/login", json={ + "username": "invalid_user", + "password": "wrong_password", + "remember_me": False + }) + + # Should fail + assert resp.status_code in [401, 403] + data = resp.json() + assert data.get("success") is not True + + +@pytest.mark.e2e +class TestCalibreWebAuthentication: + """Tests for Calibre-Web database authentication.""" + + def test_cwa_auth_mode_available(self, api_client: APIClient): + """Test that CWA auth mode is reported if configured.""" + # Check current auth mode + auth_check = api_client.get("/api/auth/check") + auth_data = auth_check.json() + + if auth_data.get("auth_mode") == "cwa": + # CWA mode is active + assert auth_data["auth_mode"] == "cwa" + # Should have authenticated or auth_required status + assert "authenticated" in auth_data + assert "auth_required" in auth_data + + +@pytest.mark.e2e +class TestAdminAccess: + """Tests for admin access restrictions.""" + + def test_settings_endpoint_respects_admin_restriction(self, api_client: APIClient): + """Test that settings endpoints respect admin restrictions.""" + # Check current auth status + auth_check = api_client.get("/api/auth/check") + auth_data = auth_check.json() + + # If auth is required and user is not admin + if auth_data.get("auth_required") and auth_data.get("authenticated"): + if not auth_data.get("is_admin"): + # Try accessing settings + resp = api_client.get("/api/settings") + + # May be blocked with 403 if admin-only + # Or allowed if settings are not restricted + assert resp.status_code in [200, 403] + + def test_onboarding_endpoint_respects_admin_restriction(self, api_client: APIClient): + """Test that onboarding endpoints respect admin restrictions.""" + # Check current auth status + auth_check = api_client.get("/api/auth/check") + auth_data = auth_check.json() + + # If auth is required and user is not admin + if auth_data.get("auth_required") and auth_data.get("authenticated"): + if not auth_data.get("is_admin"): + # Try accessing onboarding + resp = api_client.get("/api/onboarding") + + # May be blocked with 403 if admin-only + # Or allowed if settings are not restricted + assert resp.status_code in [200, 403] + + +@pytest.mark.e2e +class TestAuthenticationWorkflow: + """Tests for complete authentication workflows.""" + + def test_login_logout_cycle(self, api_client: APIClient): + """Test complete login and logout cycle.""" + # Check initial auth status + auth_check = api_client.get("/api/auth/check") + initial_auth = auth_check.json() + + # If no auth required, skip this test + if not initial_auth.get("auth_required"): + pytest.skip("No authentication required") + + # Try logout first to clear any existing session + logout_resp = api_client.post("/api/auth/logout") + assert logout_resp.status_code == 200 + + # Check we're logged out + auth_check = api_client.get("/api/auth/check") + post_logout_auth = auth_check.json() + + # For builtin/cwa auth, should not be authenticated + # For proxy auth, depends on proxy configuration + if initial_auth.get("auth_mode") in ["builtin", "cwa"]: + assert post_logout_auth.get("authenticated") is False + + def test_auth_check_consistency(self, api_client: APIClient): + """Test that auth check returns consistent results.""" + # Make multiple auth check requests + resp1 = api_client.get("/api/auth/check") + resp2 = api_client.get("/api/auth/check") + resp3 = api_client.get("/api/auth/check") + + data1 = resp1.json() + data2 = resp2.json() + data3 = resp3.json() + + # All should succeed + assert resp1.status_code == 200 + assert resp2.status_code == 200 + assert resp3.status_code == 200 + + # Auth mode should be consistent + assert data1["auth_mode"] == data2["auth_mode"] == data3["auth_mode"] + + # Auth required should be consistent + assert data1["auth_required"] == data2["auth_required"] == data3["auth_required"] diff --git a/tests/e2e/test_proxy_auth_middleware.py b/tests/e2e/test_proxy_auth_middleware.py new file mode 100644 index 00000000..939ae133 --- /dev/null +++ b/tests/e2e/test_proxy_auth_middleware.py @@ -0,0 +1,183 @@ +"""Unit tests for proxy auth middleware and admin access checks.""" + +from __future__ import annotations + +import importlib +from typing import Any +from unittest.mock import patch + +import pytest + + +def _as_response(result: Any): + if isinstance(result, tuple) and len(result) == 2: + resp, status = result + resp.status_code = status + return resp + return result + + +@pytest.fixture(scope="module") +def main_module(): + with patch("shelfmark.download.orchestrator.start"): + import shelfmark.main as main + + importlib.reload(main) + return main + + +class TestProxyAuthMiddleware: + def test_skips_for_non_proxy_mode(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + with main_module.app.test_request_context("/api/search"): + result = main_module.proxy_auth_middleware() + assert result is None + assert "user_id" not in main_module.session + + def test_skips_health_endpoint(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="proxy"): + with main_module.app.test_request_context("/api/health"): + result = main_module.proxy_auth_middleware() + assert result is None + + def test_allows_auth_check_without_header(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="proxy"): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={"PROXY_AUTH_USER_HEADER": "X-Auth-User"}, + ): + with main_module.app.test_request_context("/api/auth/check"): + result = main_module.proxy_auth_middleware() + assert result is None + assert "user_id" not in main_module.session + + def test_sets_session_from_header(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="proxy"): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={ + "PROXY_AUTH_USER_HEADER": "X-Auth-User", + "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN": False, + }, + ): + with main_module.app.test_request_context( + "/api/search", + headers={"X-Auth-User": "proxyuser"}, + ): + result = 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 + assert main_module.session.permanent is False + + def test_returns_401_when_header_missing_on_protected_path(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="proxy"): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={"PROXY_AUTH_USER_HEADER": "X-Auth-User"}, + ): + with main_module.app.test_request_context("/api/search"): + resp = _as_response(main_module.proxy_auth_middleware()) + data = resp.get_json() + + assert resp.status_code == 401 + assert "Authentication required" in (data.get("error") or "") + + def test_admin_group_membership(self, main_module): + with patch.object(main_module, "get_auth_mode", return_value="proxy"): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={ + "PROXY_AUTH_USER_HEADER": "X-Auth-User", + "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN": True, + "PROXY_AUTH_ADMIN_GROUP_HEADER": "X-Auth-Groups", + "PROXY_AUTH_ADMIN_GROUP_NAME": "admins", + }, + ): + with main_module.app.test_request_context( + "/api/search", + headers={ + "X-Auth-User": "adminuser", + "X-Auth-Groups": "users,admins,devs", + }, + ): + result = main_module.proxy_auth_middleware() + assert result is None + assert main_module.session.get("is_admin") is True + + +class TestLoginRequiredDecorator: + @pytest.fixture + def view(self): + def _view(): + return {"success": True}, 200 + + return _view + + def test_allows_no_auth(self, main_module, view): + with patch.object(main_module, "get_auth_mode", return_value="none"): + with main_module.app.test_request_context("/api/search"): + decorated = main_module.login_required(view) + resp = decorated() + + assert resp[0]["success"] is True + + def test_blocks_when_not_authenticated(self, main_module, view): + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + with main_module.app.test_request_context("/api/search"): + decorated = main_module.login_required(view) + resp = _as_response(decorated()) + + assert resp.status_code == 401 + + def test_allows_authenticated(self, main_module, view): + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + with main_module.app.test_request_context("/api/search"): + main_module.session["user_id"] = "user" + decorated = main_module.login_required(view) + resp = decorated() + + assert resp[0]["success"] is True + + def test_builtin_mode_does_not_apply_cwa_admin_setting(self, main_module, view): + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={"CWA_RESTRICT_SETTINGS_TO_ADMIN": True}, + ): + with main_module.app.test_request_context("/api/settings/general"): + main_module.session["user_id"] = "admin" + decorated = main_module.login_required(view) + resp = decorated() + + assert resp[0]["success"] is True + + def test_proxy_admin_restriction_blocks_non_admin(self, main_module, view): + with patch.object(main_module, "get_auth_mode", return_value="proxy"): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={"PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN": True}, + ): + with main_module.app.test_request_context("/api/settings/general"): + main_module.session["user_id"] = "user" + main_module.session["is_admin"] = False + decorated = main_module.login_required(view) + resp = _as_response(decorated()) + data = resp.get_json() + + assert resp.status_code == 403 + assert "Admin access required" in (data.get("error") or "") + + def test_cwa_admin_restriction_blocks_non_admin(self, main_module, view): + with patch.object(main_module, "get_auth_mode", return_value="cwa"): + with patch( + "shelfmark.core.settings_registry.load_config_file", + return_value={"CWA_RESTRICT_SETTINGS_TO_ADMIN": True}, + ): + with main_module.app.test_request_context("/api/settings/general"): + main_module.session["user_id"] = "user" + main_module.session["is_admin"] = False + decorated = main_module.login_required(view) + resp = _as_response(decorated()) + + assert resp.status_code == 403