[DOP-29772] Implement GET /v1/auth/logout endpoint

docs/changelog/next_release/275.feature.rst

@@ -0,0 +1 @@
+Implement ``GET /v1/auth/logout`` endpoint.

pyproject.toml

@@ -145,6 +145,7 @@ black = "^25.1.0"
 flake8 = "^7.2.0"
 flake8-pyproject = "^1.2.3"
 sqlalchemy = {extras = ["mypy"], version = "^2.0.40"}
+types-jwcrypto = "^1.5.0"
 
 [tool.poetry.group.docs.dependencies]
 autodoc-pydantic = "^2.2.0"

syncmaster/exceptions/auth.py

@@ -28,3 +28,19 @@ def message(self) -> str:
     @property
     def details(self) -> Any:
         return self._details
+
+
+class LogoutError(SyncmasterError):
+    """Error on logout request"""
+
+    def __init__(self, details: str) -> None:
+        self._message = "Logout error"
+        self._details = details
+
+    @property
+    def message(self) -> str:
+        return self._message
+
+    @property
+    def details(self) -> Any:
+        return self._details

syncmaster/server/api/v1/auth.py

@@ -6,6 +6,7 @@
 from fastapi import APIRouter, Depends, Request, Response
 from fastapi.security import OAuth2PasswordRequestForm
 
+from syncmaster.db.models import User
 from syncmaster.errors.registration import get_error_responses
 from syncmaster.errors.schemas.invalid_request import InvalidRequestSchema
 from syncmaster.errors.schemas.not_authorized import NotAuthorizedSchema
@@ -16,6 +17,7 @@
     DummyAuthProvider,
     KeycloakAuthProvider,
 )
+from syncmaster.server.services.get_user import get_user
 
 router = APIRouter(
     prefix="/auth",
@@ -48,9 +50,23 @@ async def auth_callback(
 ):
     token = await auth_provider.get_token_authorization_code_grant(
         code=code,
-        redirect_uri=auth_provider.settings.keycloak.redirect_uri,
     )
     request.session["access_token"] = token["access_token"]
     request.session["refresh_token"] = token["refresh_token"]
+    return Response(status_code=NO_CONTENT)
+
 
+@router.get(
+    "/logout",
+    summary="Logout user",
+    status_code=NO_CONTENT,
+)
+async def logout(
+    request: Request,
+    current_user: Annotated[User, Depends(get_user(is_active=True))],
+    auth_provider: Annotated[KeycloakAuthProvider, Depends(Stub(AuthProvider))],
+):
+    refresh_token = request.session.get("refresh_token", None)
+    request.session.clear()
+    await auth_provider.logout(user=current_user, refresh_token=refresh_token)
     return Response(status_code=NO_CONTENT)

syncmaster/server/providers/auth/base_provider.py

@@ -4,7 +4,7 @@
 from abc import ABC, abstractmethod
 from typing import Any
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 
 from syncmaster.db.models import User
 
@@ -52,7 +52,7 @@ def __init__(
         ...
 
     @abstractmethod
-    async def get_current_user(self, access_token: Any, *args, **kwargs) -> User:
+    async def get_current_user(self, access_token: Any, request: Request) -> User:
         """
         This method should return currently logged in user.
 
@@ -104,11 +104,16 @@ async def get_token_password_grant(
     async def get_token_authorization_code_grant(
         self,
         code: str,
-        redirect_uri: str,
         scopes: list[str] | None = None,
         client_id: str | None = None,
         client_secret: str | None = None,
     ) -> dict[str, Any]:
         """
         Obtain a token using the Authorization Code grant.
         """
+
+    @abstractmethod
+    async def logout(self, user: User, refresh_token: str | None) -> None:
+        """
+        Logout user
+        """

syncmaster/server/providers/auth/dummy_provider.py

@@ -20,7 +20,7 @@
 log = logging.getLogger(__name__)
 
 
-class DummyAuthProvider(AuthProvider):
+class DummyAuthProvider(AuthProvider):  # noqa: WPS338
     def __init__(
         self,
         settings: Annotated[DummyAuthProviderSettings, Depends(Stub(DummyAuthProviderSettings))],
@@ -76,16 +76,6 @@ async def get_token_password_grant(
             "expires_at": expires_at,
         }
 
-    async def get_token_authorization_code_grant(
-        self,
-        code: str,
-        redirect_uri: str,
-        scopes: list[str] | None = None,
-        client_id: str | None = None,
-        client_secret: str | None = None,
-    ) -> dict[str, Any]:
-        raise NotImplementedError("Authorization code grant is not supported by DummyAuthProvider.")
-
     def _generate_access_token(self, user_id: int) -> tuple[str, float]:
         expires_at = time() + self._settings.access_token.expire_seconds
         payload = {
@@ -109,3 +99,15 @@ def _get_user_id_from_token(self, token: str) -> int:
             return int(payload["user_id"])
         except (KeyError, TypeError, ValueError) as e:
             raise AuthorizationError("Invalid token") from e
+
+    async def get_token_authorization_code_grant(
+        self,
+        code: str,
+        scopes: list[str] | None = None,
+        client_id: str | None = None,
+        client_secret: str | None = None,
+    ) -> dict[str, Any]:
+        raise NotImplementedError("Authorization code grant is not supported by DummyAuthProvider")
+
+    async def logout(self, user: User, refresh_token: str | None) -> None:
+        raise NotImplementedError("Logout is not supported by DummyAuthProvider")

syncmaster/server/providers/auth/keycloak_provider.py

@@ -4,10 +4,12 @@
 from typing import Annotated, Any
 
 from fastapi import Depends, FastAPI, Request
-from keycloak import KeycloakOpenID
+from jwcrypto.common import JWException
+from keycloak import KeycloakOpenID, KeycloakOperationError
 
+from syncmaster.db.models.user import User
 from syncmaster.exceptions import EntityNotFoundError
-from syncmaster.exceptions.auth import AuthorizationError
+from syncmaster.exceptions.auth import AuthorizationError, LogoutError
 from syncmaster.exceptions.redirect import RedirectException
 from syncmaster.server.dependencies import Stub
 from syncmaster.server.providers.auth.base_provider import AuthProvider
@@ -55,43 +57,39 @@ async def get_token_password_grant(
     async def get_token_authorization_code_grant(
         self,
         code: str,
-        redirect_uri: str,
         scopes: list[str] | None = None,
         client_id: str | None = None,
         client_secret: str | None = None,
     ) -> dict[str, Any]:
         try:
-            redirect_uri = redirect_uri or self.settings.keycloak.redirect_uri
             token = self.keycloak_openid.token(
                 grant_type="authorization_code",
                 code=code,
-                redirect_uri=redirect_uri,
+                redirect_uri=self.settings.keycloak.redirect_uri,
             )
             return token
-        except Exception as e:
+        except KeycloakOperationError as e:
             raise AuthorizationError("Failed to get token") from e
 
-    async def get_current_user(self, access_token: str, *args, **kwargs) -> Any:  # noqa: WPS231
-        request: Request = kwargs["request"]
-        refresh_token = request.session.get("refresh_token")
-
+    async def get_current_user(self, access_token: str, request: Request) -> User:  # noqa: WPS231
         if not access_token:
             log.debug("No access token found in session.")
-            self.redirect_to_auth(request.url.path)
+            self.redirect_to_auth()
 
+        refresh_token = request.session.get("refresh_token")
         try:
             # if user is disabled or blocked in Keycloak after the token is issued, he will
             # remain authorized until the token expires (not more than 15 minutes in MTS SSO)
             token_info = self.keycloak_openid.decode_token(token=access_token)
-        except Exception as e:
+        except (KeycloakOperationError, JWException) as e:
             log.info("Access token is invalid or expired: %s", e)
             token_info = None
 
         if not token_info and refresh_token:
             log.debug("Access token invalid. Attempting to refresh.")
 
             try:
-                new_tokens = await self.refresh_access_token(refresh_token)
+                new_tokens = self.keycloak_openid.refresh_token(refresh_token)
 
                 new_access_token = new_tokens["access_token"]
                 new_refresh_token = new_tokens["refresh_token"]
@@ -102,9 +100,9 @@ async def get_current_user(self, access_token: str, *args, **kwargs) -> Any:  #
                     token=new_access_token,
                 )
                 log.debug("Access token refreshed and decoded successfully.")
-            except Exception as e:
+            except (KeycloakOperationError, JWException) as e:
                 log.debug("Failed to refresh access token: %s", e)
-                self.redirect_to_auth(request.url.path)
+                self.redirect_to_auth()
 
         if not token_info:
             raise AuthorizationError("Invalid token payload")
@@ -131,13 +129,21 @@ async def get_current_user(self, access_token: str, *args, **kwargs) -> Any:  #
                 )
         return user
 
-    async def refresh_access_token(self, refresh_token: str) -> dict[str, Any]:
-        new_tokens = self.keycloak_openid.refresh_token(refresh_token)
-        return new_tokens
-
-    def redirect_to_auth(self, path: str) -> None:
+    def redirect_to_auth(self) -> None:
         auth_url = self.keycloak_openid.auth_url(
             redirect_uri=self.settings.keycloak.redirect_uri,
             scope=self.settings.keycloak.scope,
         )
         raise RedirectException(redirect_url=auth_url)
+
+    async def logout(self, user: User, refresh_token: str | None) -> None:
+        if not refresh_token:
+            log.debug("No refresh token found in session.")
+            return
+
+        try:
+            self.keycloak_openid.logout(refresh_token)
+        except KeycloakOperationError as err:
+            msg = f"Can't logout user: {user.username}"
+            log.debug("%s. Error: %s", msg, err)
+            raise LogoutError(msg) from err

tests/test_unit/test_auth/auth_fixtures/__init__.py

@@ -1,7 +1,17 @@
 from tests.test_unit.test_auth.auth_fixtures.keycloak_fixture import (
     create_session_cookie,
+    mock_keycloak_logout,
     mock_keycloak_realm,
     mock_keycloak_token_refresh,
     mock_keycloak_well_known,
     rsa_keys,
 )
+
+__all__ = [
+    "create_session_cookie",
+    "mock_keycloak_logout",
+    "mock_keycloak_realm",
+    "mock_keycloak_token_refresh",
+    "mock_keycloak_well_known",
+    "rsa_keys",
+]