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 <git@jthoward.dev>
Co-authored-by: Alex <alex.bilbie1@gmail.com>
This commit is contained in:
Tag Howard
2026-01-15 13:27:50 +00:00
committed by GitHub
co-authored by Joshua Tag Howard Alex
parent 475ae420e5
commit 0d7a12ca7c
9 changed files with 1355 additions and 28 deletions
+10 -2
View File
@@ -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:
+137 -13
View File
@@ -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"},
),
]
+123 -12
View File
@@ -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({
+5 -1
View File
@@ -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 });
+1
View File
@@ -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
+366
View File
@@ -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"]
+253
View File
@@ -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
+277
View File
@@ -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"]
+183
View File
@@ -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