Refresh tokens, freshness pattern and scopes

fastapi_users/authentication/authenticator.py

@@ -8,6 +8,7 @@
 from fastapi_users import models
 from fastapi_users.authentication.backend import AuthenticationBackend
 from fastapi_users.authentication.strategy import Strategy
+from fastapi_users.authentication.token import UserTokenData
 from fastapi_users.manager import BaseUserManager, UserManagerDependency
 from fastapi_users.types import DependencyCallable
 
@@ -62,6 +63,7 @@ def current_user_token(
         active: bool = False,
         verified: bool = False,
         superuser: bool = False,
+        fresh: bool = False,
         get_enabled_backends: Optional[EnabledBackendsDependency] = None,
     ):
         """
@@ -89,14 +91,16 @@ def current_user_token(
 
         @with_signature(signature)
         async def current_user_token_dependency(*args, **kwargs):
-            return await self._authenticate(
+            token_data, token = await self._authenticate(
                 *args,
                 optional=optional,
                 active=active,
                 verified=verified,
                 superuser=superuser,
+                fresh=fresh,
                 **kwargs,
             )
+            return token_data.user, token
 
         return current_user_token_dependency
 
@@ -106,6 +110,7 @@ def current_user(
         active: bool = False,
         verified: bool = False,
         superuser: bool = False,
+        fresh: bool = False,
         get_enabled_backends: Optional[EnabledBackendsDependency] = None,
     ):
         """
@@ -133,18 +138,68 @@ def current_user(
 
         @with_signature(signature)
         async def current_user_dependency(*args, **kwargs):
-            user, _ = await self._authenticate(
+            token_data, _ = await self._authenticate(
                 *args,
                 optional=optional,
                 active=active,
                 verified=verified,
                 superuser=superuser,
+                fresh=fresh,
                 **kwargs,
             )
-            return user
+            if token_data:
+                return token_data.user
+            return None
 
         return current_user_dependency
 
+    def current_token(
+        self,
+        optional: bool = False,
+        active: bool = False,
+        verified: bool = False,
+        superuser: bool = False,
+        fresh: bool = False,
+        get_enabled_backends: Optional[EnabledBackendsDependency] = None,
+    ):
+        """
+        Return a dependency callable to retrieve the full token data for the currently authenticated user.
+
+        :param optional: If `True`, `None` is returned if there is no authenticated user
+        or if it doesn't pass the other requirements.
+        Otherwise, throw `401 Unauthorized`. Defaults to `False`.
+        Otherwise, an exception is raised. Defaults to `False`.
+        :param active: If `True`, throw `401 Unauthorized` if
+        the authenticated user is inactive. Defaults to `False`.
+        :param verified: If `True`, throw `401 Unauthorized` if
+        the authenticated user is not verified. Defaults to `False`.
+        :param superuser: If `True`, throw `403 Forbidden` if
+        the authenticated user is not a superuser. Defaults to `False`.
+        :param get_enabled_backends: Optional dependency callable returning
+        a list of enabled authentication backends.
+        Useful if you want to dynamically enable some authentication backends
+        based on external logic, like a configuration in database.
+        By default, all specified authentication backends are enabled.
+        Please not however that every backends will appear in the OpenAPI documentation,
+        as FastAPI resolves it statically.
+        """
+        signature = self._get_dependency_signature(get_enabled_backends)
+
+        @with_signature(signature)
+        async def current_token_dependency(*args, **kwargs):
+            token_data, _ = await self._authenticate(
+                *args,
+                optional=optional,
+                active=active,
+                verified=verified,
+                superuser=superuser,
+                fresh=fresh,
+                **kwargs,
+            )
+            return token_data
+
+        return current_token_dependency
+
     async def _authenticate(
         self,
         *args,
@@ -153,37 +208,41 @@ async def _authenticate(
         active: bool = False,
         verified: bool = False,
         superuser: bool = False,
+        fresh: bool = False,
         **kwargs,
-    ) -> Tuple[Optional[models.UP], Optional[str]]:
-        user: Optional[models.UP] = None
+    ) -> Tuple[Optional[UserTokenData[models.UP, models.ID]], Optional[str]]:
+        token_data: Optional[UserTokenData[models.UP, models.ID]] = None
         token: Optional[str] = None
         enabled_backends: Sequence[AuthenticationBackend] = kwargs.get(
             "enabled_backends", self.backends
         )
         for backend in self.backends:
             if backend in enabled_backends:
                 token = kwargs[name_to_variable_name(backend.name)]
-                strategy: Strategy[models.UP, models.ID] = kwargs[
+                strategy: Strategy = kwargs[
                     name_to_strategy_variable_name(backend.name)
                 ]
                 if token is not None:
-                    user = await strategy.read_token(token, user_manager)
-                    if user:
+                    token_data = await strategy.read_token(token, user_manager)
+                    if token_data:
                         break
 
         status_code = status.HTTP_401_UNAUTHORIZED
-        if user:
-            status_code = status.HTTP_403_FORBIDDEN
-            if active and not user.is_active:
-                status_code = status.HTTP_401_UNAUTHORIZED
-                user = None
-            elif (
-                verified and not user.is_verified or superuser and not user.is_superuser
-            ):
-                user = None
-        if not user and not optional:
+        if token_data:
+            if token_data.user:
+                status_code = status.HTTP_403_FORBIDDEN
+                if active and not token_data.user.is_active:
+                    status_code = status.HTTP_401_UNAUTHORIZED
+                    token_data = None
+                elif (
+                    (verified and not token_data.user.is_verified)
+                    or (superuser and not token_data.user.is_superuser)
+                    or (fresh and not token_data.fresh)
+                ):
+                    token_data = None
+        if not token_data and not optional:
             raise HTTPException(status_code=status_code)
-        return user, token
+        return token_data, token
 
     def _get_dependency_signature(
         self, get_enabled_backends: Optional[EnabledBackendsDependency] = None

fastapi_users/authentication/backend.py

@@ -1,4 +1,5 @@
-from typing import Any, Generic
+from datetime import datetime
+from typing import Any, Generic, Optional, Set
 
 from fastapi import Response
 
@@ -7,14 +8,19 @@
     Strategy,
     StrategyDestroyNotSupportedError,
 )
+from fastapi_users.authentication.token import UserTokenData
 from fastapi_users.authentication.transport import (
+    LoginT,
+    LogoutT,
     Transport,
     TransportLogoutNotSupportedError,
+    TransportTokenResponse,
 )
+from fastapi_users.scopes import SystemScope
 from fastapi_users.types import DependencyCallable
 
 
-class AuthenticationBackend(Generic[models.UP, models.ID]):
+class AuthenticationBackend(Generic[LoginT, LogoutT]):
     """
     Combination of an authentication transport and strategy.
 
@@ -27,34 +33,67 @@ class AuthenticationBackend(Generic[models.UP, models.ID]):
     """
 
     name: str
-    transport: Transport
+    transport: Transport[LoginT, LogoutT]
 
     def __init__(
         self,
         name: str,
-        transport: Transport,
-        get_strategy: DependencyCallable[Strategy[models.UP, models.ID]],
+        transport: Transport[LoginT, LogoutT],
+        get_strategy: DependencyCallable[Strategy],
+        access_token_lifetime_seconds: Optional[int] = 3600,
+        refresh_token_enabled: bool = False,
+        refresh_token_lifetime_seconds: Optional[int] = 86400,
     ):
         self.name = name
         self.transport = transport
         self.get_strategy = get_strategy
+        self.access_token_lifetime_seconds = access_token_lifetime_seconds
+        self.refresh_token_enabled = refresh_token_enabled
+        self.refresh_token_lifetime_seconds = refresh_token_lifetime_seconds
 
     async def login(
         self,
-        strategy: Strategy[models.UP, models.ID],
-        user: models.UP,
+        strategy: Strategy,
+        user: models.UserProtocol[Any],
         response: Response,
-    ) -> Any:
-        token = await strategy.write_token(user)
-        return await self.transport.get_login_response(token, response)
+        last_authenticated: Optional[datetime] = None,
+    ) -> Optional[LoginT]:
+        scopes: Set[str] = set()
+        if user.is_active:
+            scopes.add(SystemScope.USER)
+        if user.is_verified:
+            scopes.add(SystemScope.VERIFIED)
+        if user.is_superuser:
+            scopes.add(SystemScope.SUPERUSER)
+
+        access_token_data = UserTokenData.issue_now(
+            user,
+            self.access_token_lifetime_seconds,
+            last_authenticated,
+            scopes=scopes,
+        )
+        token_response = TransportTokenResponse(
+            access_token=await strategy.write_token(access_token_data)
+        )
+        if self.refresh_token_enabled:
+            refresh_token_data = UserTokenData.issue_now(
+                user,
+                self.refresh_token_lifetime_seconds,
+                last_authenticated,
+                scopes={SystemScope.REFRESH},
+            )
+            token_response.refresh_token = await strategy.write_token(
+                refresh_token_data
+            )
+        return await self.transport.get_login_response(token_response, response)
 
     async def logout(
         self,
-        strategy: Strategy[models.UP, models.ID],
-        user: models.UP,
+        strategy: Strategy,
+        user: models.UserProtocol[Any],
         token: str,
         response: Response,
-    ) -> Any:
+    ) -> Optional[LogoutT]:
         try:
             await strategy.destroy_token(token, user)
         except StrategyDestroyNotSupportedError:

fastapi_users/authentication/strategy/base.py

@@ -1,27 +1,37 @@
 import sys
-from typing import Generic, Optional
+from typing import Any, Dict, Generic, Optional
 
 if sys.version_info < (3, 8):
     from typing_extensions import Protocol  # pragma: no cover
 else:
     from typing import Protocol  # pragma: no cover
 
 from fastapi_users import models
+from fastapi_users.authentication.token import UserTokenData
 from fastapi_users.manager import BaseUserManager
 
 
 class StrategyDestroyNotSupportedError(Exception):
     pass
 
 
-class Strategy(Protocol, Generic[models.UP, models.ID]):
+class Strategy(Protocol):
     async def read_token(
-        self, token: Optional[str], user_manager: BaseUserManager[models.UP, models.ID]
-    ) -> Optional[models.UP]:
+        self,
+        token: Optional[str],
+        user_manager: BaseUserManager[models.UP, models.ID],
+    ) -> Optional[UserTokenData[models.UP, models.ID]]:
         ...  # pragma: no cover
 
-    async def write_token(self, user: models.UP) -> str:
+    async def write_token(
+        self,
+        token_data: UserTokenData[models.UserProtocol[Any], Any],
+    ) -> str:
         ...  # pragma: no cover
 
-    async def destroy_token(self, token: str, user: models.UP) -> None:
+    async def destroy_token(
+        self,
+        token: str,
+        user: models.UserProtocol[Any],
+    ) -> None:
         ...  # pragma: no cover

fastapi_users/authentication/strategy/db/models.py

@@ -1,6 +1,6 @@
 import sys
 from datetime import datetime
-from typing import TypeVar
+from typing import Optional, TypeVar
 
 if sys.version_info < (3, 8):
     from typing_extensions import Protocol  # pragma: no cover
@@ -16,6 +16,9 @@ class AccessTokenProtocol(Protocol[models.ID]):
     token: str
     user_id: models.ID
     created_at: datetime
+    expires_at: Optional[datetime]
+    last_authenticated: datetime
+    scopes: str
 
     def __init__(self, *args, **kwargs) -> None:
         ...  # pragma: no cover