Fix the metrics endpoint for SSO auth

Author: uittenbroekrobbert

operations-manager/python/opi/middleware/authorization.py

@@ -11,7 +11,7 @@
 
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.requests import Request
-from starlette.responses import RedirectResponse, Response
+from starlette.responses import JSONResponse, RedirectResponse, Response
 from starlette.routing import Match
 
 from opi.services.user_service import get_user_service
@@ -21,10 +21,14 @@
 
 logger = logging.getLogger(__name__)
 
-# URL prefixes that skip authorization checks and logging (health checks, metrics, etc.)
+# Exact paths that skip authorization checks. These MUST be exact matches and not
+# prefixes, otherwise they would also match longer authenticated routes (e.g. the
+# "/metrics" Prometheus scrape endpoint must not match the "/metrics-explorer" page).
+SKIP_AUTH_EXACT = ("/metrics",)
+
+# URL prefixes that skip authorization checks and logging (health checks, static assets).
 SKIP_AUTH_PREFIXES = (
     "/health",
-    "/metrics",
     "/ready",
     "/static/",
 )
@@ -79,13 +83,23 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -
         path = request.url.path
 
         # Skip auth checks and logging for infrastructure endpoints
-        if path.startswith(SKIP_AUTH_PREFIXES):
+        if path in SKIP_AUTH_EXACT or path.startswith(SKIP_AUTH_PREFIXES):
             return await call_next(request)
 
-        # API routes should use API key authentication, not SSO by default
+        # API routes use API key authentication by default, but honor an explicit
+        # @requires_sso marking (e.g. browser-called endpoints like metrics-explorer).
+        # These get the same session gate as web pages, but with JSON responses.
         if path.startswith("/api/"):
-            # For API routes, only require SSO if explicitly marked with @requires_sso
-            request.state.user = None  # API routes don't use session-based user
+            user = get_user(request)
+            request.state.user = user
+            if self._route_requires_sso(request):
+                if not user:
+                    logger.info(f"Rejecting unauthenticated SSO-required API request: {path}")
+                    return JSONResponse(status_code=401, content={"error": "Authentication required"})
+                user_email = user.get("email")
+                if user_email and not get_user_service().is_email_allowed(user_email):
+                    logger.warning(f"Access denied for user {user_email} on {path} - not in allowlist")
+                    return JSONResponse(status_code=403, content={"error": "Access denied"})
             return await call_next(request)
 
         # Get user from session

operations-manager/python/tests/test_authorization.py

@@ -80,18 +80,100 @@ async def test_static_files_always_pass(self):
         call_next.assert_awaited_once_with(request)
 
     @pytest.mark.asyncio
-    async def test_api_routes_skip_sso(self):
-        """API routes should skip SSO and set user to None."""
+    async def test_api_routes_skip_sso_by_default(self):
+        """API routes without @requires_sso should skip SSO and set user to None."""
         middleware = self._make_middleware()
         request = self._make_request("/api/v1/projects")
         expected_response = Response(content="ok")
         call_next = AsyncMock(return_value=expected_response)
 
-        result = await middleware.dispatch(request, call_next)
+        with patch.object(middleware, "_route_requires_sso", return_value=False):
+            result = await middleware.dispatch(request, call_next)
 
         assert result is expected_response
         assert request.state.user is None
 
+    @pytest.mark.asyncio
+    async def test_api_route_requires_sso_rejects_unauthenticated(self):
+        """API route marked @requires_sso must return 401 when there is no session user."""
+        middleware = self._make_middleware()
+        request = self._make_request("/api/metrics-explorer/metrics/minio", session={})
+        call_next = AsyncMock()
+
+        with patch.object(middleware, "_route_requires_sso", return_value=True):
+            result = await middleware.dispatch(request, call_next)
+
+        assert result.status_code == 401
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_api_route_requires_sso_allows_authenticated(self):
+        """API route marked @requires_sso should pass through for an allowlisted session user."""
+        middleware = self._make_middleware()
+        user = {"email": "test@example.com"}
+        request = self._make_request("/api/metrics-explorer/metrics/minio", session={"user": user})
+        expected_response = Response(content="ok")
+        call_next = AsyncMock(return_value=expected_response)
+
+        mock_user_service = MagicMock()
+        mock_user_service.is_email_allowed.return_value = True
+
+        with (
+            patch.object(middleware, "_route_requires_sso", return_value=True),
+            patch("opi.middleware.authorization.get_user_service", return_value=mock_user_service),
+        ):
+            result = await middleware.dispatch(request, call_next)
+
+        assert result is expected_response
+        assert request.state.user == user
+
+    @pytest.mark.asyncio
+    async def test_api_route_requires_sso_blocks_non_allowlisted(self):
+        """API route marked @requires_sso must return 403 for a non-allowlisted user."""
+        middleware = self._make_middleware()
+        user = {"email": "blocked@example.com"}
+        request = self._make_request("/api/metrics-explorer/metrics/minio", session={"user": user})
+        call_next = AsyncMock()
+
+        mock_user_service = MagicMock()
+        mock_user_service.is_email_allowed.return_value = False
+
+        with (
+            patch.object(middleware, "_route_requires_sso", return_value=True),
+            patch("opi.middleware.authorization.get_user_service", return_value=mock_user_service),
+        ):
+            result = await middleware.dispatch(request, call_next)
+
+        assert result.status_code == 403
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_metrics_scrape_endpoint_skips_auth(self):
+        """The exact /metrics Prometheus scrape endpoint must remain public."""
+        middleware = self._make_middleware()
+        request = self._make_request("/metrics")
+        expected_response = Response(content="ok")
+        call_next = AsyncMock(return_value=expected_response)
+
+        result = await middleware.dispatch(request, call_next)
+
+        assert result is expected_response
+        call_next.assert_awaited_once_with(request)
+
+    @pytest.mark.asyncio
+    async def test_metrics_explorer_is_not_skipped_by_metrics_prefix(self):
+        """Regression: /metrics-explorer must NOT be treated as public by the /metrics rule."""
+        middleware = self._make_middleware()
+        request = self._make_request("/metrics-explorer", session={})
+        call_next = AsyncMock()
+
+        with patch.object(middleware, "_route_requires_sso", return_value=True):
+            result = await middleware.dispatch(request, call_next)
+
+        assert result.status_code == 302
+        assert result.headers.get("location") == "/auth/login"
+        call_next.assert_not_awaited()
+
     @pytest.mark.asyncio
     async def test_unauthenticated_user_redirected(self):
         """Should redirect to /auth/login when SSO required and no user."""