Ability to use token in noop auth (if needed)

ols/src/auth/auth_dependency_interface.py

@@ -2,7 +2,33 @@
 
 from abc import ABC, abstractmethod
 
-from fastapi import Request
+from fastapi import HTTPException, Request
+
+
+def extract_bearer_token(header: str) -> str:
+    """Extract the bearer token from an HTTP authorization header.
+
+    Args:
+        header: The authorization header containing the token.
+
+    Returns:
+        The extracted token if present, else an empty string.
+    """
+    try:
+        scheme, token = header.split(" ", 1)
+        return token if scheme.lower() == "bearer" else ""
+    except ValueError:
+        return ""
+
+
+def extract_token_from_request(request: Request) -> str:
+    """Extract the bearer token from a FastAPI request."""
+    authorization_header = request.headers.get("Authorization")
+    if not authorization_header:
+        raise HTTPException(
+            status_code=401, detail="Unauthorized: No auth header found"
+        )
+    return extract_bearer_token(authorization_header)
 
 
 class AuthDependencyInterface(ABC):

ols/src/auth/k8s.py

@@ -17,7 +17,10 @@
     RUNNING_IN_CLUSTER,
 )
 
-from .auth_dependency_interface import AuthDependencyInterface
+from .auth_dependency_interface import (
+    AuthDependencyInterface,
+    extract_token_from_request,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -209,22 +212,6 @@ def get_user_info(token: str) -> Optional[kubernetes.client.V1TokenReview]:
         ) from e
 
 
-def _extract_bearer_token(header: str) -> str:
-    """Extract the bearer token from an HTTP authorization header.
-
-    Args:
-        header: The authorization header containing the token.
-
-    Returns:
-        The extracted token if present, else an empty string.
-    """
-    try:
-        scheme, token = header.split(" ", 1)
-        return token if scheme.lower() == "bearer" else ""
-    except ValueError:
-        return ""
-
-
 class AuthDependency(AuthDependencyInterface):
     """Create an AuthDependency Class that allows customizing the acces Scope path to check."""
 
@@ -263,12 +250,7 @@ async def __call__(self, request: Request) -> tuple[str, str, bool, str]:
             # conversation history and user feedback depend on having any
             # user ID (identity) in proper format (UUID)
             return DEFAULT_USER_UID, DEFAULT_USER_NAME, False, NO_USER_TOKEN
-        authorization_header = request.headers.get("Authorization")
-        if not authorization_header:
-            raise HTTPException(
-                status_code=401, detail="Unauthorized: No auth header found"
-            )
-        token = _extract_bearer_token(authorization_header)
+        token = extract_token_from_request(request)
         if not token:
             raise HTTPException(
                 status_code=401,

ols/src/auth/noop.py

@@ -2,10 +2,11 @@
 
 import logging
 
-from fastapi import Request
+from fastapi import HTTPException, Request
 
 from ols import config
 from ols.constants import DEFAULT_USER_NAME, DEFAULT_USER_UID, NO_USER_TOKEN
+from ols.src.auth.auth_dependency_interface import extract_token_from_request
 
 from .auth_dependency_interface import AuthDependencyInterface
 
@@ -33,7 +34,12 @@ async def __call__(self, request: Request) -> tuple[str, str, bool, str]:
             If user_id check should be skipped
             User's token
         """
-        user_token = NO_USER_TOKEN  # no-op auth yield no token
+        # the user token is optional in the noop auth, if there is no token we
+        # use our default value for this scenario
+        try:
+            user_token = extract_token_from_request(request)
+        except HTTPException:
+            user_token = NO_USER_TOKEN
         if config.dev_config.disable_auth:
             if (
                 config.ols_config.logging_config is None
@@ -58,4 +64,9 @@ async def __call__(self, request: Request) -> tuple[str, str, bool, str]:
         # try to read user ID from request
         user_id = request.query_params.get("user_id", DEFAULT_USER_UID)
         logger.info("User ID: %s", user_id)
-        return user_id, DEFAULT_USER_NAME, self.skip_userid_check, user_token
+        return (
+            user_id,
+            DEFAULT_USER_NAME,
+            self.skip_userid_check,
+            user_token,
+        )

tests/unit/auth/test_auth.py

@@ -1,10 +1,15 @@
 """Unit tests for auth/auth module."""
 
 import pytest
+from fastapi import HTTPException, Request
 
 from ols.app.models.config import AuthenticationConfig, OLSConfig
 from ols.src.auth import k8s, noop
 from ols.src.auth.auth import get_auth_dependency, use_k8s_auth
+from ols.src.auth.auth_dependency_interface import (
+    extract_bearer_token,
+    extract_token_from_request,
+)
 
 
 def test_use_k8s_auth_no_auth_config():
@@ -78,3 +83,24 @@ def test_get_auth_dependency_unknown_module():
         Exception, match="Invalid/unknown auth. module was configured: foo"
     ):
         get_auth_dependency(ols_config, "/path")
+
+
+def test_extract_bearer_token():
+    """Test the function extract_bearer_token."""
+    # good value
+    assert extract_bearer_token("Bearer token") == "token"
+
+    # bad values
+    assert extract_bearer_token("Bearer") == ""
+    assert extract_bearer_token("sha256~aaaaaa") == ""
+
+
+def test_extract_token_from_request():
+    """Test the function extract_token_from_request."""
+    request = Request(
+        scope={"type": "http", "headers": [(b"authorization", b"Bearer token")]}
+    )
+    assert extract_token_from_request(request) == "token"
+
+    with pytest.raises(HTTPException, match="Unauthorized: No auth header found"):
+        extract_token_from_request(Request(scope={"type": "http", "headers": []}))

tests/unit/auth/test_noop_auth_dependency.py

@@ -70,3 +70,26 @@ async def test_noop_auth_dependency_call_with_user_id():
     assert username == DEFAULT_USER_NAME
     assert skip_user_id_check is True
     assert token == NO_USER_TOKEN
+
+
+@pytest.mark.asyncio
+async def test_noop_auth_dependency_provided_token():
+    """Check that the no-op auth. dependency returns provided token."""
+    path = "/ols-access"
+    auth_dependency = noop.AuthDependency(virtual_path=path)
+    # Simulate a request with user_id specified as optional parameter
+    user_id_in_request = "00000000-1234-1234-1234-000000000000"
+    request = Request(
+        scope={
+            "type": "http",
+            "headers": [(b"authorization", b"Bearer user-token")],
+            "query_string": f"user_id={user_id_in_request}",
+        }
+    )
+    user_uid, username, skip_user_id_check, token = await auth_dependency(request)
+
+    # Check if the correct user info has been returned
+    assert user_uid == user_id_in_request
+    assert username == DEFAULT_USER_NAME
+    assert skip_user_id_check is True
+    assert token == "user-token"  # noqa: S105