diff --git a/app/auth.py b/app/auth.py index d478d83d..32e86001 100644 --- a/app/auth.py +++ b/app/auth.py @@ -66,105 +66,110 @@ def get_gravatar_url(email): return f"https://www.gravatar.com/avatar/{email_hash}?d=identicon" -if AUTH_ENABLED: +async def login(request: Request): + """Show login page with appropriate authentication options""" + return templates.TemplateResponse( + "login.html", + { + "request": request, + "error": request.query_params.get("error"), + "message": request.query_params.get("message"), + "show_oauth": OAUTH_CONFIGURED, + "oauth_provider_name": OAUTH_PROVIDER_NAME, + "app_version": settings.version, # Changed from app_version to version + }, + ) - @router.get("/login") - async def login(request: Request): - """Show login page with appropriate authentication options""" - return templates.TemplateResponse( - "login.html", - { - "request": request, - "error": request.query_params.get("error"), - "message": request.query_params.get("message"), - "show_oauth": OAUTH_CONFIGURED, - "oauth_provider_name": OAUTH_PROVIDER_NAME, - "app_version": settings.version, # Changed from app_version to version - }, - ) - @router.get("/oauth-login") - async def oauth_login(request: Request): - """Handle OAuth login flow""" - if not OAUTH_CONFIGURED: - return RedirectResponse(url="/login?error=OAuth+not+configured", status_code=status.HTTP_302_FOUND) +async def oauth_login(request: Request): + """Handle OAuth login flow""" + if not OAUTH_CONFIGURED: + return RedirectResponse(url="/login?error=OAuth+not+configured", status_code=status.HTTP_302_FOUND) - redirect_uri = request.url_for("oauth_callback") - return await oauth.authentik.authorize_redirect(request, redirect_uri) + redirect_uri = request.url_for("oauth_callback") + return await oauth.authentik.authorize_redirect(request, redirect_uri) - @router.get("/oauth-callback") - async def oauth_callback(request: Request): - """Handle OAuth callback from provider""" - try: - token = await oauth.authentik.authorize_access_token(request) - userinfo = token.get("userinfo") - if not userinfo: - return RedirectResponse( - url="/login?error=Failed+to+retrieve+user+information", status_code=status.HTTP_302_FOUND - ) - # Store user info in session - user_data = dict(userinfo) - - # Add Gravatar picture if no picture is provided - if not user_data.get("picture") and user_data.get("email"): - user_data["picture"] = get_gravatar_url(user_data["email"]) - - # Check if user is admin based on OAuth groups or specific email - # You can customize this logic based on your OAuth provider's attributes - # For example, check if user has an "admin" group or specific email domain - is_admin = False - if "groups" in user_data: - # Check if user is in admin group - groups = user_data.get("groups", []) - admin_group = (settings.admin_group_name or "admin").strip().lower() - is_admin = admin_group in [group.lower() for group in groups] - - # Set is_admin flag (defaults to False for OAuth users unless they're in admin group) - user_data["is_admin"] = is_admin - - request.session["user"] = user_data - - # Log the successful authentication - print(f"User authenticated via OAuth: {user_data.get('email', 'No email')} (admin: {is_admin})") - - # Redirect to original destination or default - redirect_url = request.session.pop("redirect_after_login", "/upload") - return RedirectResponse(url=redirect_url) - except Exception as e: - print(f"OAuth authentication error: {str(e)}") +async def oauth_callback(request: Request): + """Handle OAuth callback from provider""" + try: + token = await oauth.authentik.authorize_access_token(request) + userinfo = token.get("userinfo") + if not userinfo: return RedirectResponse( - url=f"/login?error=Authentication+failed:+{str(e)}", status_code=status.HTTP_302_FOUND + url="/login?error=Failed+to+retrieve+user+information", status_code=status.HTTP_302_FOUND ) - @router.post("/auth") - async def auth(request: Request): - """Handle local username/password authentication""" - form_data = await request.form() - username = form_data.get("username") - password = form_data.get("password") + # Store user info in session + user_data = dict(userinfo) - if username == settings.admin_username and password == settings.admin_password: - # Create user session - request.session["user"] = { - "id": "admin", - "name": "Administrator", - "email": f"{username}@local.docuelevate", - "preferred_username": username, - "picture": "/static/images/default-avatar.svg", - "is_admin": True, - } - # Redirect to original destination or default - redirect_url = request.session.pop("redirect_after_login", "/upload") - return RedirectResponse(url=redirect_url, status_code=302) - else: - return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302) + # Add Gravatar picture if no picture is provided + if not user_data.get("picture") and user_data.get("email"): + user_data["picture"] = get_gravatar_url(user_data["email"]) - @router.get("/logout") - async def logout(request: Request): - """Handle user logout""" - request.session.pop("user", None) - return RedirectResponse(url="/login?message=You+have+been+logged+out+successfully", status_code=302) + # Check if user is admin based on OAuth groups or specific email + # You can customize this logic based on your OAuth provider's attributes + # For example, check if user has an "admin" group or specific email domain + is_admin = False + if "groups" in user_data: + # Check if user is in admin group + groups = user_data.get("groups", []) + admin_group = (settings.admin_group_name or "admin").strip().lower() + is_admin = admin_group in [group.lower() for group in groups] + + # Set is_admin flag (defaults to False for OAuth users unless they're in admin group) + user_data["is_admin"] = is_admin + + request.session["user"] = user_data + + # Log the successful authentication + print(f"User authenticated via OAuth: {user_data.get('email', 'No email')} (admin: {is_admin})") + + # Redirect to original destination or default + redirect_url = request.session.pop("redirect_after_login", "/upload") + return RedirectResponse(url=redirect_url) + except Exception as e: + print(f"OAuth authentication error: {str(e)}") + return RedirectResponse( + url=f"/login?error=Authentication+failed:+{str(e)}", status_code=status.HTTP_302_FOUND + ) + + +async def auth(request: Request): + """Handle local username/password authentication""" + form_data = await request.form() + username = form_data.get("username") + password = form_data.get("password") + + if username == settings.admin_username and password == settings.admin_password: + # Create user session + request.session["user"] = { + "id": "admin", + "name": "Administrator", + "email": f"{username}@local.docuelevate", + "preferred_username": username, + "picture": "/static/images/default-avatar.svg", + "is_admin": True, + } + # Redirect to original destination or default + redirect_url = request.session.pop("redirect_after_login", "/upload") + return RedirectResponse(url=redirect_url, status_code=302) + else: + return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302) + + +async def logout(request: Request): + """Handle user logout""" + request.session.pop("user", None) + return RedirectResponse(url="/login?message=You+have+been+logged+out+successfully", status_code=302) + + +if AUTH_ENABLED: + router.add_api_route("/login", login, methods=["GET"]) + router.add_api_route("/oauth-login", oauth_login, methods=["GET"]) + router.add_api_route("/oauth-callback", oauth_callback, methods=["GET"]) + router.add_api_route("/auth", auth, methods=["POST"]) + router.add_api_route("/logout", logout, methods=["GET"]) @router.get("/api/auth/whoami") diff --git a/tests/conftest_oauth.py b/tests/conftest_oauth.py index 58b619f3..b152d552 100644 --- a/tests/conftest_oauth.py +++ b/tests/conftest_oauth.py @@ -205,8 +205,8 @@ def oauth_enabled_app(oauth_config: Dict[str, str]): from fastapi.testclient import TestClient from app.main import app - # Create test client - client = TestClient(app) + # Create test client with base_url to satisfy TrustedHostMiddleware + client = TestClient(app, base_url="http://localhost") yield client diff --git a/tests/test_auth_module.py b/tests/test_auth_module.py index 486a7a7c..3dbcd98d 100644 --- a/tests/test_auth_module.py +++ b/tests/test_auth_module.py @@ -163,7 +163,7 @@ class TestRequireLogin: mock_request.session = {"user": {"id": "123"}} mock_request.url = "http://localhost/test" - result = test_sync_endpoint(mock_request) + result = await test_sync_endpoint(mock_request) assert result == {"message": "success"} @@ -180,7 +180,7 @@ class TestRequireLogin: mock_request.session = {} mock_request.url = "http://localhost/protected" - result = test_sync_endpoint(mock_request) + result = await test_sync_endpoint(mock_request) assert isinstance(result, RedirectResponse) assert result.status_code == status.HTTP_302_FOUND @@ -192,43 +192,42 @@ class TestOAuthConfiguration: def test_oauth_not_configured_without_credentials(self): """Test OAuth is not configured when credentials are missing.""" - with patch("app.auth.settings") as mock_settings: - mock_settings.auth_enabled = True - mock_settings.authentik_client_id = None - mock_settings.authentik_client_secret = None + import importlib - # Re-import to trigger configuration logic - import importlib + import app.auth - import app.auth + try: + with patch("app.config.settings") as mock_settings: + mock_settings.auth_enabled = True + mock_settings.authentik_client_id = None + mock_settings.authentik_client_secret = None + importlib.reload(app.auth) + + assert app.auth.OAUTH_CONFIGURED is False + finally: importlib.reload(app.auth) - from app.auth import OAUTH_CONFIGURED - - assert OAUTH_CONFIGURED is False - def test_oauth_configured_with_credentials(self): """Test OAuth is configured when credentials are provided.""" - with patch("app.auth.settings") as mock_settings: - with patch("app.auth.oauth") as mock_oauth: + import importlib + + import app.auth + + try: + with patch("app.config.settings") as mock_settings: mock_settings.auth_enabled = True mock_settings.authentik_client_id = "test_client_id" mock_settings.authentik_client_secret = "test_secret" mock_settings.authentik_config_url = "https://auth.example.com/.well-known/openid-configuration" mock_settings.oauth_provider_name = "Test SSO" - # Re-import to trigger configuration logic - import importlib - - import app.auth - importlib.reload(app.auth) - from app.auth import OAUTH_CONFIGURED, OAUTH_PROVIDER_NAME - - assert OAUTH_CONFIGURED is True - assert OAUTH_PROVIDER_NAME == "Test SSO" + assert app.auth.OAUTH_CONFIGURED is True + assert app.auth.OAUTH_PROVIDER_NAME == "Test SSO" + finally: + importlib.reload(app.auth) @pytest.mark.unit diff --git a/tests/test_oauth_integration_flows.py b/tests/test_oauth_integration_flows.py index 3739f4d2..4d0bc1ad 100644 --- a/tests/test_oauth_integration_flows.py +++ b/tests/test_oauth_integration_flows.py @@ -64,6 +64,7 @@ class TestOAuthLoginFlow: class TestOAuthCallback: """Test OAuth callback handling.""" + @pytest.mark.asyncio @patch("app.auth.oauth.authentik.authorize_access_token") async def test_oauth_callback_with_valid_token( self, mock_authorize, oauth_enabled_app: TestClient, test_user_info: dict @@ -86,6 +87,7 @@ class TestOAuthCallback: # Should redirect after successful login assert response.status_code == 302 + @pytest.mark.asyncio @patch("app.auth.oauth.authentik.authorize_access_token") async def test_oauth_callback_stores_user_in_session( self, mock_authorize, oauth_enabled_app: TestClient, test_user_info: dict @@ -108,6 +110,7 @@ class TestOAuthCallback: # Should set session cookie assert "set-cookie" in response.headers or response.status_code == 302 + @pytest.mark.asyncio @patch("app.auth.oauth.authentik.authorize_access_token") async def test_oauth_callback_with_admin_user( self, mock_authorize, oauth_enabled_app: TestClient @@ -131,6 +134,7 @@ class TestOAuthCallback: # Should successfully authenticate assert response.status_code == 302 + @pytest.mark.asyncio @patch("app.auth.oauth.authentik.authorize_access_token") async def test_oauth_callback_rejects_non_admin( self, mock_authorize, oauth_enabled_app: TestClient @@ -160,6 +164,7 @@ class TestOAuthCallback: class TestOAuthSessionManagement: """Test session management with OAuth authentication.""" + @pytest.mark.asyncio @patch("app.auth.oauth.authentik.authorize_access_token") async def test_authenticated_user_can_access_protected_routes( self, mock_authorize, oauth_enabled_app: TestClient, test_user_info: dict @@ -192,6 +197,7 @@ class TestOAuthSessionManagement: if response.status_code == 302: assert "/login" in response.headers.get("location", "") + @pytest.mark.asyncio @patch("app.auth.oauth.authentik.authorize_access_token") async def test_logout_clears_session( self, mock_authorize, oauth_enabled_app: TestClient, test_user_info: dict @@ -226,6 +232,7 @@ class TestOAuthErrorHandling: # Should handle error gracefully assert response.status_code in [302, 400] + @pytest.mark.asyncio @patch("app.auth.oauth.authentik.authorize_access_token") async def test_oauth_callback_with_invalid_token( self, mock_authorize, oauth_enabled_app: TestClient @@ -244,6 +251,7 @@ class TestOAuthErrorHandling: location = response.headers.get("location", "") assert "error" in location.lower() or "login" in location.lower() + @pytest.mark.asyncio @patch("app.auth.oauth.authentik.authorize_access_token") async def test_oauth_callback_without_userinfo( self, mock_authorize, oauth_enabled_app: TestClient diff --git a/tests/test_rate_limit_decorators.py b/tests/test_rate_limit_decorators.py index 6e023683..761ffcfa 100644 --- a/tests/test_rate_limit_decorators.py +++ b/tests/test_rate_limit_decorators.py @@ -136,7 +136,7 @@ class TestRateLimitDecorators: # Mock limiter to return a simple passthrough decorator mock_limiter = MagicMock() - mock_limiter.exempt.return_value = lambda f: f + mock_limiter.exempt.side_effect = lambda f: f mock_get_limiter.return_value = mock_limiter # Decorate function diff --git a/tests/test_step_timeout.py b/tests/test_step_timeout.py index f4b7a374..bc6eaa5b 100644 --- a/tests/test_step_timeout.py +++ b/tests/test_step_timeout.py @@ -44,8 +44,8 @@ class TestStepTimeout: # Mock database session with proper query chain mock_db = MagicMock() - # Set up the query chain to return empty list - mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [] + # Set up the query chain to return empty list (single .filter() call with multiple conditions) + mock_db.query.return_value.filter.return_value.all.return_value = [] # Run function count = mark_stalled_steps_as_failed(mock_db) @@ -74,8 +74,8 @@ class TestStepTimeout: # Mock database session mock_db = MagicMock() - # Set up the query chain to return stalled steps - mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [step1, step2] + # Set up the query chain to return stalled steps (single .filter() call with multiple conditions) + mock_db.query.return_value.filter.return_value.all.return_value = [step1, step2] # Run function count = mark_stalled_steps_as_failed(mock_db) @@ -105,8 +105,8 @@ class TestStepTimeout: # Mock database session mock_db = MagicMock() - # Set up the query chain to return stalled step - mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [step] + # Set up the query chain to return stalled step (single .filter() call with multiple conditions) + mock_db.query.return_value.filter.return_value.all.return_value = [step] # Run function with 150 second timeout count = mark_stalled_steps_as_failed(mock_db, timeout_seconds=150) @@ -131,9 +131,8 @@ class TestStepTimeout: # Mock database session with file filter mock_db = MagicMock() - # Set up the query chain with file filter - mock_query_chain = mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value - mock_query_chain.filter.return_value.all.return_value = [step] + # Set up the query chain with file filter (first .filter() for conditions, second for file_id) + mock_db.query.return_value.filter.return_value.filter.return_value.all.return_value = [step] # Run function for specific file count = mark_stalled_steps_as_failed(mock_db, file_id=42) @@ -158,8 +157,8 @@ class TestStepTimeout: # Mock database session mock_db = MagicMock() - # Set up the query chain to return stalled step - mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [step] + # Set up the query chain to return stalled step (single .filter() call with multiple conditions) + mock_db.query.return_value.filter.return_value.all.return_value = [step] # Run function count = mark_stalled_steps_as_failed(mock_db, timeout_seconds=600) @@ -186,8 +185,8 @@ class TestStepTimeout: # Mock database session mock_db = MagicMock() - # Set up the query chain to return stalled step - mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [step] + # Set up the query chain to return stalled step (single .filter() call with multiple conditions) + mock_db.query.return_value.filter.return_value.all.return_value = [step] # Run function count = mark_stalled_steps_as_failed(mock_db) @@ -212,9 +211,8 @@ class TestStepTimeout: # Mock database session mock_db = MagicMock() - # Set up the query chain with file filter - mock_query_chain = mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value - mock_query_chain.filter.return_value.all.return_value = [step] + # Set up the query chain with file filter (first .filter() for conditions, second for file_id) + mock_db.query.return_value.filter.return_value.filter.return_value.all.return_value = [step] # Run function result = check_and_recover_stalled_file(mock_db, 42) @@ -229,9 +227,8 @@ class TestStepTimeout: # Mock database session with no stalled steps mock_db = MagicMock() - # Set up the query chain with file filter - mock_query_chain = mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value - mock_query_chain.filter.return_value.all.return_value = [] + # Set up the query chain with file filter (first .filter() for conditions, second for file_id) + mock_db.query.return_value.filter.return_value.filter.return_value.all.return_value = [] # Run function result = check_and_recover_stalled_file(mock_db, 42)