fief_client.integrations.fastapi

  1"""FastAPI integration."""
  2
  3import uuid
  4from collections.abc import AsyncGenerator, Coroutine, Generator
  5from inspect import Parameter, Signature, isawaitable
  6from typing import (
  7    Callable,
  8    Optional,
  9    Protocol,
 10    TypeVar,
 11    Union,
 12    cast,
 13)
 14
 15from fastapi import Depends, HTTPException, Request, Response, status
 16from fastapi.security.base import SecurityBase
 17from fastapi.security.http import HTTPAuthorizationCredentials
 18from makefun import with_signature
 19
 20from fief_client import (
 21    Fief,
 22    FiefAccessTokenACRTooLow,
 23    FiefAccessTokenExpired,
 24    FiefAccessTokenInfo,
 25    FiefAccessTokenInvalid,
 26    FiefAccessTokenMissingPermission,
 27    FiefAccessTokenMissingScope,
 28    FiefACR,
 29    FiefAsync,
 30    FiefUserInfo,
 31)
 32
 33FiefClientClass = Union[Fief, FiefAsync]
 34
 35TokenType = Union[str, HTTPAuthorizationCredentials]
 36
 37RETURN_TYPE = TypeVar("RETURN_TYPE")
 38
 39DependencyCallable = Callable[
 40    ...,
 41    Union[
 42        RETURN_TYPE,
 43        Coroutine[None, None, RETURN_TYPE],
 44        AsyncGenerator[RETURN_TYPE, None],
 45        Generator[RETURN_TYPE, None, None],
 46    ],
 47]
 48
 49
 50class UserInfoCacheProtocol(Protocol):
 51    """
 52    Protocol that should follow a class to implement a cache mechanism for user information.
 53
 54    Read more: https://docs.fief.dev/integrate/python/fastapi/#caching-user-information
 55    """
 56
 57    async def get(self, user_id: uuid.UUID) -> Optional[FiefUserInfo]:
 58        """
 59        Retrieve user information from cache, if available.
 60
 61        :param user_id: The ID of the user to retrieve information for.
 62        """
 63        ...  # pragma: no cover
 64
 65    async def set(self, user_id: uuid.UUID, userinfo: FiefUserInfo) -> None:
 66        """
 67        Store user information in cache.
 68
 69        :param user_id: The ID of the user to cache information for.
 70        :param userinfo: The user information to cache.
 71        """
 72        ...  # pragma: no cover
 73
 74
 75class FiefAuth:
 76    """
 77    Helper class to integrate Fief authentication with FastAPI.
 78
 79    **Example:**
 80
 81    ```py
 82    from fastapi.security import OAuth2AuthorizationCodeBearer
 83    from fief_client import FiefAccessTokenInfo, FiefAsync
 84    from fief_client.integrations.fastapi import FiefAuth
 85
 86    fief = FiefAsync(
 87        "https://example.fief.dev",
 88        "YOUR_CLIENT_ID",
 89        "YOUR_CLIENT_SECRET",
 90    )
 91
 92    scheme = OAuth2AuthorizationCodeBearer(
 93        "https://example.fief.dev/authorize",
 94        "https://example.fief.dev/api/token",
 95        scopes={"openid": "openid", "offline_access": "offline_access"},
 96    )
 97
 98    auth = FiefAuth(fief, scheme)
 99    ```
100    """
101
102    def __init__(
103        self,
104        client: FiefClientClass,
105        scheme: SecurityBase,
106        *,
107        get_userinfo_cache: Optional[DependencyCallable[UserInfoCacheProtocol]] = None,
108    ) -> None:
109        """
110        :param client: Instance of a Fief client.
111        Can be either `fief_client.Fief` or `fief_client.FiefAsync`.
112        :param scheme: FastAPI security scheme.
113        It'll be used to retrieve the access token in the request.
114        :param get_userinfo_cache: Optional dependency returning an instance of a class
115        following the `UserInfoCacheProtocol`.
116        It'll be used to cache user information on your server.
117        Otherwise, the Fief API will always be reached when requesting user information.
118        """
119        self.client = client
120        self.scheme = scheme
121        self.get_userinfo_cache = get_userinfo_cache
122
123    def authenticated(
124        self,
125        optional: bool = False,
126        scope: Optional[list[str]] = None,
127        acr: Optional[FiefACR] = None,
128        permissions: Optional[list[str]] = None,
129    ):
130        """
131        Return a FastAPI dependency to check if a request is authenticated.
132
133        If the request is authenticated, the dependency will return a `fief_client.FiefAccessTokenInfo`.
134
135        :param optional: If `False` and the request is not authenticated,
136        an unauthorized response will be raised.
137        :param scope: Optional list of scopes required.
138        If the access token lacks one of the required scope, a forbidden response will be raised.
139        :param acr: Optional minimum ACR level required.
140        If the access token doesn't meet the minimum level, a forbidden response will be raised.
141        Read more: https://docs.fief.dev/going-further/acr/
142        :param permissions: Optional list of permissions required.
143        If the access token lacks one of the required permission, a forbidden response will be raised.
144
145        **Example**
146
147        ```py
148        @app.get("/authenticated")
149        async def get_authenticated(
150            access_token_info: FiefAccessTokenInfo = Depends(auth.authenticated()),
151        ):
152            return access_token_info
153        ```
154        """
155        signature = self._get_authenticated_call_signature(self.scheme)
156
157        @with_signature(signature)
158        async def _authenticated(
159            request: Request, response: Response, token: Optional[TokenType]
160        ) -> Optional[FiefAccessTokenInfo]:
161            if token is None:
162                if optional:
163                    return None
164                return await self.get_unauthorized_response(request, response)
165
166            if isinstance(token, HTTPAuthorizationCredentials):
167                token = token.credentials
168
169            try:
170                result = self.client.validate_access_token(
171                    token,
172                    required_scope=scope,
173                    required_acr=acr,
174                    required_permissions=permissions,
175                )
176                if isawaitable(result):
177                    info = await result
178                else:
179                    info = result
180            except (FiefAccessTokenInvalid, FiefAccessTokenExpired):
181                if optional:
182                    return None
183                return await self.get_unauthorized_response(request, response)
184            except (
185                FiefAccessTokenMissingScope,
186                FiefAccessTokenACRTooLow,
187                FiefAccessTokenMissingPermission,
188            ):
189                return await self.get_forbidden_response(request, response)
190
191            return info
192
193        return _authenticated
194
195    def current_user(
196        self,
197        optional: bool = False,
198        scope: Optional[list[str]] = None,
199        acr: Optional[FiefACR] = None,
200        permissions: Optional[list[str]] = None,
201        refresh: bool = False,
202    ):
203        """
204        Return a FastAPI dependency to check if a user is authenticated.
205
206        If the request is authenticated, the dependency will return a `fief_client.FiefUserInfo`.
207
208        If provided, the cache mechanism will be used to retrieve this information without calling the Fief API.
209
210        :param optional: If `False` and the request is not authenticated,
211        an unauthorized response will be raised.
212        :param scope: Optional list of scopes required.
213        If the access token lacks one of the required scope, a forbidden response will be raised.
214        :param acr: Optional minimum ACR level required.
215        If the access token doesn't meet the minimum level, a forbidden response will be raised.
216        Read more: https://docs.fief.dev/going-further/acr/
217        :param permissions: Optional list of permissions required.
218        If the access token lacks one of the required permission, a forbidden response will be raised.
219        :param refresh: If `True`, the user information will be refreshed from the Fief API.
220        Otherwise, the cache will be used.
221
222        **Example**
223
224        ```py
225        @app.get("/current-user", name="current_user")
226        async def get_current_user(
227            user: FiefUserInfo = Depends(auth.current_user()),
228        ):
229            return {"email": user["email"]}
230        ```
231        """
232        signature = self._get_current_user_call_signature(
233            self.authenticated(optional, scope, acr, permissions)
234        )
235
236        @with_signature(signature)
237        async def _current_user(
238            access_token_info: Optional[FiefAccessTokenInfo], *args, **kwargs
239        ) -> Optional[FiefUserInfo]:
240            userinfo_cache: Optional[UserInfoCacheProtocol] = kwargs.get(
241                "userinfo_cache"
242            )
243
244            if access_token_info is None and optional:
245                return None
246            assert access_token_info is not None
247
248            userinfo = None
249            if userinfo_cache is not None:
250                userinfo = await userinfo_cache.get(access_token_info["id"])
251
252            if userinfo is None or refresh:
253                result = self.client.userinfo(access_token_info["access_token"])
254                if isawaitable(result):
255                    userinfo = cast(FiefUserInfo, await result)
256                else:
257                    userinfo = cast(FiefUserInfo, result)
258
259                if userinfo_cache is not None:
260                    await userinfo_cache.set(access_token_info["id"], userinfo)
261
262            return userinfo
263
264        return _current_user
265
266    async def get_unauthorized_response(self, request: Request, response: Response):
267        """
268        Raise an `fastapi.HTTPException` with the status code 401.
269
270        This method is called when using the `authenticated` or `current_user` dependency
271        but the request is not authenticated.
272
273        You can override this method to customize the behavior in this case.
274        """
275        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
276
277    async def get_forbidden_response(self, request: Request, response: Response):
278        """
279        Raise an `fastapi.HTTPException` with the status code 403.
280
281        This method is called when using the `authenticated` or `current_user` dependency
282        but the access token doesn't match the list of scopes or permissions.
283
284        You can override this method to customize the behavior in this case.
285        """
286        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
287
288    def _get_authenticated_call_signature(self, scheme: SecurityBase) -> Signature:
289        """
290        Generate a dynamic signature for the authenticated dependency.
291        Here comes some blood magic 🧙‍♂️
292        Thank to "makefun", we are able to generate callable
293        with a dynamic security scheme dependency at runtime.
294        This way, it's detected by the OpenAPI generator.
295        """
296        parameters: list[Parameter] = [
297            Parameter(
298                name="request",
299                kind=Parameter.POSITIONAL_OR_KEYWORD,
300                annotation=Request,
301            ),
302            Parameter(
303                name="response",
304                kind=Parameter.POSITIONAL_OR_KEYWORD,
305                annotation=Response,
306            ),
307            Parameter(
308                name="token",
309                kind=Parameter.POSITIONAL_OR_KEYWORD,
310                default=Depends(cast(Callable, scheme)),
311            ),
312        ]
313
314        return Signature(parameters)
315
316    def _get_current_user_call_signature(self, authenticated: Callable) -> Signature:
317        """
318        Generate a dynamic signature for the current_user dependency.
319        Here comes some blood magic 🧙‍♂️
320        Thank to "makefun", we are able to generate callable
321        with a dynamic security scheme dependency at runtime.
322        This way, it's detected by the OpenAPI generator.
323        """
324        parameters: list[Parameter] = [
325            Parameter(
326                name="access_token_info",
327                kind=Parameter.POSITIONAL_OR_KEYWORD,
328                default=Depends(authenticated),
329                annotation=FiefAccessTokenInfo,
330            ),
331        ]
332
333        if self.get_userinfo_cache is not None:
334            parameters.append(
335                Parameter(
336                    name="userinfo_cache",
337                    kind=Parameter.POSITIONAL_OR_KEYWORD,
338                    default=Depends(self.get_userinfo_cache),
339                    annotation=UserInfoCacheProtocol,
340                ),
341            )
342
343        return Signature(parameters)
344
345
346__all__ = ["FiefAuth", "FiefClientClass", "UserInfoCacheProtocol"]