HTTPBearer token is set, Auth button not shown on /api/docs

.github/workflows/test-package.yml

@@ -46,7 +46,7 @@ jobs:
         env:
           COVERAGE_FILE: reports/.coverage.${{ matrix.python-version }}
       - name: Upload pytest test results
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v4
         with:
           name: reports
           path: reports
@@ -61,7 +61,7 @@ jobs:
         with:
           python-version: '3.8'
       - name: Get coverage files
-        uses: actions/download-artifact@v3
+        uses: actions/download-artifact@v4
         with:
           name: reports
           path: reports

oauth2_lib/fastapi.py

@@ -155,6 +155,49 @@ async def extract(self, request: Request) -> str | None:
         return credential.credentials if credential else None
 
 
+class TokenExtractor(HTTPBearer):
+    """Extracts tokens from HTTP requests.
+
+    Specifically designed for HTTP Authorization header token extraction.
+    """
+
+    def __init__(self, auto_error: bool = False):
+        super().__init__(scheme_name="Token", auto_error=auto_error)
+
+    async def __call__(self, request: Request, token: str | None = None) -> str | None:
+        """Extract the token from the request.
+
+        Args:
+            request: Fastapi Request object.
+            token: token
+
+        Returns:
+            str: The token extracted from the request.
+
+        """
+        # Handle WebSocket requests separately only to check for token presence.
+        if isinstance(request, WebSocket):
+            if token is None:
+                raise HTTPException(
+                    status_code=HTTPStatus.FORBIDDEN,
+                    detail="Not authenticated",
+                )
+            token_or_extracted_id_token = token
+        else:
+            request = cast(Request, request)
+
+            if token is None:
+                credentials = await super().__call__(request)
+                if not credentials:
+                    return None
+
+                token_or_extracted_id_token = credentials.credentials
+            else:
+                token_or_extracted_id_token = token
+
+        return token_or_extracted_id_token
+
+
 class OIDCAuth(Authentication):
     """Implements OIDC authentication.
 
@@ -181,14 +224,17 @@ def __init__(
 
         self.openid_config: OIDCConfig | None = None
 
-    async def authenticate(self, request: HTTPConnection, token: str | None = None) -> OIDCUserModel | None:
+    async def authenticate(
+        self, request: Request, token: str | None = None, is_strawberry_request: bool = False
+    ) -> OIDCUserModel | None:
         """Return the OIDC user from OIDC introspect endpoint.
 
         This is used as a security module in Fastapi projects
 
         Args:
             request: Starlette request/websocket method.
             token: Optional value to directly pass a token.
+            is_strawberry_request: argument to signify strawberry request.
 
         Returns:
             OIDCUserModel object.
@@ -197,33 +243,19 @@ async def authenticate(self, request: HTTPConnection, token: str | None = None)
         if not oauth2lib_settings.OAUTH2_ACTIVE:
             return None
 
-        async with AsyncClient(http1=True, verify=HTTPX_SSL_CONTEXT) as async_client:
-            await self.check_openid_config(async_client)
-
-            # Handle WebSocket requests separately only to check for token presence.
-            if isinstance(request, WebSocket):
-                if token is None:
-                    raise HTTPException(
-                        status_code=HTTPStatus.FORBIDDEN,
-                        detail="Not authenticated",
-                    )
-                token_or_extracted_id_token = token
-            else:
-                request = cast(Request, request)
+        if await self.is_bypassable_request(request):
+            return None
 
-                if await self.is_bypassable_request(request):
-                    return None
+        if is_strawberry_request:
+            token = await self.id_token_extractor.extract(request)
 
-                if token is None:
-                    extracted_id_token = await self.id_token_extractor.extract(request)
-                    if not extracted_id_token:
-                        raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail="Not authenticated")
+        if not token:
+            raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail="Not authenticated")
 
-                    token_or_extracted_id_token = extracted_id_token
-                else:
-                    token_or_extracted_id_token = token
+        async with AsyncClient(http1=True, verify=HTTPX_SSL_CONTEXT) as async_client:
+            await self.check_openid_config(async_client)
 
-            user_info: OIDCUserModel = await self.userinfo(async_client, token_or_extracted_id_token)
+            user_info: OIDCUserModel = await self.userinfo(async_client, token)
             logger.debug("OIDCUserModel object.", user_info=user_info)
             return user_info
 

oauth2_lib/strawberry.py

@@ -56,7 +56,7 @@ async def get_current_user(self) -> OIDCUserModel | None:
             return None
 
         try:
-            return await self.auth_manager.authentication.authenticate(self.request)
+            return await self.auth_manager.authentication.authenticate(self.request, is_strawberry_request=True)
         except HTTPException as exc:
             logger.debug("User is not authenticated", status_code=exc.status_code, detail=exc.detail)
             return None

tests/test_fastapi.py

@@ -8,7 +8,7 @@
 from httpx import AsyncClient, BasicAuth
 from starlette.websockets import WebSocket
 
-from oauth2_lib.fastapi import HttpBearerExtractor, OIDCAuth, OIDCConfig, OIDCUserModel
+from oauth2_lib.fastapi import HttpBearerExtractor, OIDCAuth, OIDCConfig, OIDCUserModel, TokenExtractor
 from oauth2_lib.settings import oauth2lib_settings
 from tests.conftest import MockResponse
 
@@ -165,7 +165,8 @@ async def test_authenticate_success(make_mock_async_client, discovery, oidc_auth
         request = mock.MagicMock(spec=Request)
         request.headers = {"Authorization": "Bearer valid_token"}
 
-        user = await oidc_auth.authenticate(request)
+        token = await TokenExtractor().__call__(request)
+        user = await oidc_auth.authenticate(request, token)
         assert user == user_info_matching, "Authentication failed for a valid token"