feat:use id token for linkedin userinfo (#229)

* feat:use id token for linkedin userinfo

+ remove Python 3.8 support

* chore: implement missing tests

* chore: reorganize imports in test files for clarity

Author: tomasvotava

.github/workflows/lint.yml

@@ -8,7 +8,6 @@ jobs:
     strategy:
       matrix:
         python-version:
-          - "3.8"
           - "3.9"
           - "3.10"
           - "3.11"

.github/workflows/test.yml

@@ -11,7 +11,6 @@ jobs:
     strategy:
       matrix:
         python-version:
-          - "3.8"
           - "3.9"
           - "3.10"
           - "3.11"

examples/generic.py

@@ -1,6 +1,6 @@
 """This is an example usage of fastapi-sso."""
 
-from typing import Any, Dict, Union
+from typing import Any, Union
 from httpx import AsyncClient
 import uvicorn
 from fastapi import FastAPI, HTTPException
@@ -23,7 +23,7 @@
 # and then python examples/generic.py
 
 
-def convert_openid(response: Dict[str, Any], _client: Union[AsyncClient, None]) -> OpenID:
+def convert_openid(response: dict[str, Any], _client: Union[AsyncClient, None]) -> OpenID:
     """Convert user information returned by OIDC"""
     print(response)
     return OpenID(display_name=response["sub"])

examples/linkedin.py

@@ -13,7 +13,7 @@
 sso = LinkedInSSO(
     client_id=CLIENT_ID,
     client_secret=CLIENT_SECRET,
-    redirect_uri="http://localhost:5000/auth/callback",
+    redirect_uri="http://localhost:5050/auth/callback",
     allow_insecure_http=True,
 )
 
@@ -34,4 +34,4 @@ async def auth_callback(request: Request):
 
 
 if __name__ == "__main__":
-    uvicorn.run(app="examples.linkedin:app", host="127.0.0.1", port=5000)
+    uvicorn.run(app="examples.linkedin:app", host="127.0.0.1", port=5050)

fastapi_sso/pkce.py

@@ -3,7 +3,6 @@
 import base64
 import hashlib
 import os
-from typing import Tuple
 
 
 def get_code_verifier(length: int = 96) -> str:
@@ -13,7 +12,7 @@ def get_code_verifier(length: int = 96) -> str:
     return base64.urlsafe_b64encode(os.urandom(bytes_length)).decode("utf-8").replace("=", "")[:length]
 
 
-def get_pkce_challenge_pair(verifier_length: int = 96) -> Tuple[str, str]:
+def get_pkce_challenge_pair(verifier_length: int = 96) -> tuple[str, str]:
     """Get tuple of (verifier, challenge) for PKCE challenge."""
     code_verifier = get_code_verifier(verifier_length)
     code_challenge = (

fastapi_sso/sso/base.py

@@ -7,9 +7,10 @@
 import sys
 import warnings
 from types import TracebackType
-from typing import Any, ClassVar, Dict, List, Literal, Optional, Type, TypedDict, TypeVar, Union, overload
+from typing import Any, ClassVar, Literal, Optional, TypedDict, TypeVar, Union, overload
 
 import httpx
+import jwt
 import pydantic
 from oauthlib.oauth2 import WebApplicationClient
 from starlette.exceptions import HTTPException
@@ -33,6 +34,10 @@
 P = ParamSpec("P")
 
 
+def _decode_id_token(id_token: str, verify: bool = False) -> dict:
+    return jwt.decode(id_token, options={"verify_signature": verify})
+
+
 class DiscoveryDocument(TypedDict):
     """Discovery document."""
 
@@ -95,10 +100,11 @@ class SSOBase:
     client_id: str = NotImplemented
     client_secret: str = NotImplemented
     redirect_uri: Optional[Union[pydantic.AnyHttpUrl, str]] = NotImplemented
-    scope: ClassVar[List[str]] = []
-    additional_headers: ClassVar[Optional[Dict[str, Any]]] = None
+    scope: ClassVar[list[str]] = []
+    additional_headers: ClassVar[Optional[dict[str, Any]]] = None
     uses_pkce: bool = False
     requires_state: bool = False
+    use_id_token_for_user_info: ClassVar[bool] = False
 
     _pkce_challenge_length: int = 96
 
@@ -109,7 +115,7 @@ def __init__(
         redirect_uri: Optional[Union[pydantic.AnyHttpUrl, str]] = None,
         allow_insecure_http: bool = False,
         use_state: bool = False,
-        scope: Optional[List[str]] = None,
+        scope: Optional[list[str]] = None,
     ):
         """Base class (mixin) for all SSO providers."""
         self.client_id: str = client_id
@@ -224,6 +230,18 @@ async def openid_from_response(self, response: dict, session: Optional[httpx.Asy
         """
         raise NotImplementedError(f"Provider {self.provider} not supported")
 
+    async def openid_from_token(self, id_token: dict, session: Optional[httpx.AsyncClient] = None) -> OpenID:
+        """Converts an ID token from the provider's token endpoint to an OpenID object.
+
+        Args:
+            id_token (dict): The id token data retrieved from the token endpoint.
+            session: (Optional[httpx.AsyncClient]): The HTTPX AsyncClient session.
+
+        Returns:
+            OpenID: The user information in a standardized format.
+        """
+        raise NotImplementedError(f"Provider {self.provider} not supported")
+
     async def get_discovery_document(self) -> DiscoveryDocument:
         """Retrieves the discovery document containing useful URLs.
 
@@ -257,14 +275,14 @@ async def get_login_url(
         self,
         *,
         redirect_uri: Optional[Union[pydantic.AnyHttpUrl, str]] = None,
-        params: Optional[Dict[str, Any]] = None,
+        params: Optional[dict[str, Any]] = None,
         state: Optional[str] = None,
     ) -> str:
         """Generates and returns the prepared login URL.
 
         Args:
             redirect_uri (Optional[str]): Overrides the `redirect_uri` specified on this instance.
-            params (Optional[Dict[str, Any]]): Additional query parameters to add to the login request.
+            params (Optional[dict[str, Any]]): Additional query parameters to add to the login request.
             state (Optional[str]): The state parameter for the OAuth 2.0 authorization request.
 
         Raises:
@@ -304,14 +322,14 @@ async def get_login_redirect(
         self,
         *,
         redirect_uri: Optional[str] = None,
-        params: Optional[Dict[str, Any]] = None,
+        params: Optional[dict[str, Any]] = None,
         state: Optional[str] = None,
     ) -> RedirectResponse:
         """Constructs and returns a redirect response to the login page of OAuth SSO provider.
 
         Args:
             redirect_uri (Optional[str]): Overrides the `redirect_uri` specified on this instance.
-            params (Optional[Dict[str, Any]]): Additional query parameters to add to the login request.
+            params (Optional[dict[str, Any]]): Additional query parameters to add to the login request.
             state (Optional[str]): The state parameter for the OAuth 2.0 authorization request.
 
         Returns:
@@ -330,8 +348,8 @@ async def verify_and_process(
         self,
         request: Request,
         *,
-        params: Optional[Dict[str, Any]] = None,
-        headers: Optional[Dict[str, Any]] = None,
+        params: Optional[dict[str, Any]] = None,
+        headers: Optional[dict[str, Any]] = None,
         redirect_uri: Optional[str] = None,
         convert_response: Literal[True] = True,
     ) -> Optional[OpenID]: ...
@@ -341,28 +359,28 @@ async def verify_and_process(
         self,
         request: Request,
         *,
-        params: Optional[Dict[str, Any]] = None,
-        headers: Optional[Dict[str, Any]] = None,
+        params: Optional[dict[str, Any]] = None,
+        headers: Optional[dict[str, Any]] = None,
         redirect_uri: Optional[str] = None,
         convert_response: Literal[False],
-    ) -> Optional[Dict[str, Any]]: ...
+    ) -> Optional[dict[str, Any]]: ...
 
     @requires_async_context
     async def verify_and_process(
         self,
         request: Request,
         *,
-        params: Optional[Dict[str, Any]] = None,
-        headers: Optional[Dict[str, Any]] = None,
+        params: Optional[dict[str, Any]] = None,
+        headers: Optional[dict[str, Any]] = None,
         redirect_uri: Optional[str] = None,
         convert_response: Union[Literal[True], Literal[False]] = True,
-    ) -> Union[Optional[OpenID], Optional[Dict[str, Any]]]:
+    ) -> Union[Optional[OpenID], Optional[dict[str, Any]]]:
         """Processes the login given a FastAPI (Starlette) Request object. This should be used for the /callback path.
 
         Args:
             request (Request): FastAPI or Starlette request object.
-            params (Optional[Dict[str, Any]]): Additional query parameters to pass to the provider.
-            headers (Optional[Dict[str, Any]]): Additional headers to pass to the provider.
+            params (Optional[dict[str, Any]]): Additional query parameters to pass to the provider.
+            headers (Optional[dict[str, Any]]): Additional headers to pass to the provider.
             redirect_uri (Optional[str]): Overrides the `redirect_uri` specified on this instance.
             convert_response (bool): If True, userinfo response is converted to OpenID object.
 
@@ -371,7 +389,7 @@ async def verify_and_process(
 
         Returns:
             Optional[OpenID]: User information as OpenID instance (if convert_response == True)
-            Optional[Dict[str, Any]]: The original JSON response from the API.
+            Optional[dict[str, Any]]: The original JSON response from the API.
         """
         headers = headers or {}
         code = request.query_params.get("code")
@@ -433,7 +451,7 @@ async def __aenter__(self) -> "SSOBase":
 
     async def __aexit__(
         self,
-        _exc_type: Optional[Type[BaseException]],
+        _exc_type: Optional[type[BaseException]],
         _exc_val: Optional[BaseException],
         _exc_tb: Optional[TracebackType],
     ) -> None:
@@ -442,14 +460,14 @@ async def __aexit__(
 
     def __exit__(
         self,
-        _exc_type: Optional[Type[BaseException]],
+        _exc_type: Optional[type[BaseException]],
         _exc_val: Optional[BaseException],
         _exc_tb: Optional[TracebackType],
     ) -> None:
         return None
 
     @property
-    def _extra_query_params(self) -> Dict:
+    def _extra_query_params(self) -> dict:
         return {}
 
     @overload
@@ -458,8 +476,8 @@ async def process_login(
         code: str,
         request: Request,
         *,
-        params: Optional[Dict[str, Any]] = None,
-        additional_headers: Optional[Dict[str, Any]] = None,
+        params: Optional[dict[str, Any]] = None,
+        additional_headers: Optional[dict[str, Any]] = None,
         redirect_uri: Optional[str] = None,
         pkce_code_verifier: Optional[str] = None,
         convert_response: Literal[True] = True,
@@ -471,33 +489,33 @@ async def process_login(
         code: str,
         request: Request,
         *,
-        params: Optional[Dict[str, Any]] = None,
-        additional_headers: Optional[Dict[str, Any]] = None,
+        params: Optional[dict[str, Any]] = None,
+        additional_headers: Optional[dict[str, Any]] = None,
         redirect_uri: Optional[str] = None,
         pkce_code_verifier: Optional[str] = None,
         convert_response: Literal[False],
-    ) -> Optional[Dict[str, Any]]: ...
+    ) -> Optional[dict[str, Any]]: ...
 
     @requires_async_context
     async def process_login(
         self,
         code: str,
         request: Request,
         *,
-        params: Optional[Dict[str, Any]] = None,
-        additional_headers: Optional[Dict[str, Any]] = None,
+        params: Optional[dict[str, Any]] = None,
+        additional_headers: Optional[dict[str, Any]] = None,
         redirect_uri: Optional[str] = None,
         pkce_code_verifier: Optional[str] = None,
         convert_response: Union[Literal[True], Literal[False]] = True,
-    ) -> Union[Optional[OpenID], Optional[Dict[str, Any]]]:
+    ) -> Union[Optional[OpenID], Optional[dict[str, Any]]]:
         """Processes login from the callback endpoint to verify the user and request user info endpoint.
         It's a lower-level method, typically, you should use `verify_and_process` instead.
 
         Args:
             code (str): The authorization code.
             request (Request): FastAPI or Starlette request object.
-            params (Optional[Dict[str, Any]]): Additional query parameters to pass to the provider.
-            additional_headers (Optional[Dict[str, Any]]): Additional headers to be added to all requests.
+            params (Optional[dict[str, Any]]): Additional query parameters to pass to the provider.
+            additional_headers (Optional[dict[str, Any]]): Additional headers to be added to all requests.
             redirect_uri (Optional[str]): Overrides the `redirect_uri` specified on this instance.
             pkce_code_verifier (Optional[str]): A PKCE code verifier sent to the server to verify the login request.
             convert_response (bool): If True, userinfo response is converted to OpenID object.
@@ -507,7 +525,7 @@ async def process_login(
 
         Returns:
             Optional[OpenID]: User information in OpenID format if the login was successful (convert_response == True).
-            Optional[Dict[str, Any]]: Original userinfo API endpoint response.
+            Optional[dict[str, Any]]: Original userinfo API endpoint response.
         """
         if self._oauth_client is not None:  # pragma: no cover
             self._oauth_client = None
@@ -565,5 +583,9 @@ async def process_login(
             response = await session.get(uri)
             content = response.json()
             if convert_response:
+                if self.use_id_token_for_user_info:
+                    if not self._id_token:
+                        raise SSOLoginError(401, f"Provider {self.provider!r} did not return id token.")
+                    return await self.openid_from_token(_decode_id_token(self._id_token), session)
                 return await self.openid_from_response(content, session)
             return content

fastapi_sso/sso/bitbucket.py

@@ -1,6 +1,6 @@
 """BitBucket SSO Oauth Helper class"""
 
-from typing import TYPE_CHECKING, ClassVar, List, Optional, Union
+from typing import TYPE_CHECKING, ClassVar, Optional, Union
 
 import pydantic
 
@@ -23,7 +23,7 @@ def __init__(
         client_secret: str,
         redirect_uri: Optional[Union[pydantic.AnyHttpUrl, str]] = None,
         allow_insecure_http: bool = False,
-        scope: Optional[List[str]] = None,
+        scope: Optional[list[str]] = None,
     ):
         super().__init__(
             client_id=client_id,

fastapi_sso/sso/discord.py

@@ -1,6 +1,6 @@
 """Discord SSO Oauth Helper class"""
 
-from typing import TYPE_CHECKING, ClassVar, List, Optional, Union
+from typing import TYPE_CHECKING, ClassVar, Optional, Union
 
 import pydantic
 
@@ -22,7 +22,7 @@ def __init__(
         client_secret: str,
         redirect_uri: Optional[Union[pydantic.AnyHttpUrl, str]] = None,
         allow_insecure_http: bool = False,
-        scope: Optional[List[str]] = None,
+        scope: Optional[list[str]] = None,
     ):
         super().__init__(
             client_id=client_id,

fastapi_sso/sso/generic.py

@@ -3,7 +3,7 @@
 """
 
 import logging
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Type, Union
+from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
 from fastapi_sso.sso.base import DiscoveryDocument, OpenID, SSOBase
 
@@ -16,10 +16,10 @@
 def create_provider(
     *,
     name: str = "generic",
-    default_scope: Optional[List[str]] = None,
+    default_scope: Optional[list[str]] = None,
     discovery_document: Union[DiscoveryDocument, Callable[[SSOBase], DiscoveryDocument]],
-    response_convertor: Optional[Callable[[Dict[str, Any], Optional["httpx.AsyncClient"]], OpenID]] = None
-) -> Type[SSOBase]:
+    response_convertor: Optional[Callable[[dict[str, Any], Optional["httpx.AsyncClient"]], OpenID]] = None
+) -> type[SSOBase]:
     """A factory to create a generic OAuth client usable with almost any OAuth provider.
     Returns a class.