✨(backend) integrate Keycloak JWT support

Added support for Keycloak JWT tokens,
including custom claims and a dedicated `KeycloakAccessToken` class.

Author: kernicPanel

CHANGELOG.md

@@ -23,6 +23,7 @@ and this project adheres to
   in admin backoffice
 - Add batch order export csv for admin backoffice
 - Add quote definitions admin in the back office
+- Add support for keycloak openid token
 
 ### Fixed
 

env.d/development/common.dist

@@ -19,6 +19,12 @@ JOANIE_LMS_BACKENDS = '[{
 
 #JWT
 DJANGO_JWT_PRIVATE_SIGNING_KEY=ThisIsAnExampleKeyForDevPurposeOnly
+DJANGO_JWT_ALGORITHM=RS256
+DJANGO_JWT_TYPE_CLAIM=typ
+DJANGO_KEYCLOAK_USERNAME_CLAIM=preferred_username
+DJANGO_KEYCLOAK_ISSUER=http://keycloak/realms/master
+DJANGO_KEYCLOAK_JWK_URL=http://keycloak/realms/master/protocol/openid-connect/certs
+JOANIE_JWT_USER_FIELDS_SYNC={"username": "preferred_username", "first_name": "given_name", "last_name": "family_name", "email": "email", "language": "locale"}
 
 # Joanie settings
 

src/backend/joanie/core/authentication.py

@@ -11,9 +11,10 @@
 from rest_framework_simplejwt.authentication import JWTAuthentication
 from rest_framework_simplejwt.exceptions import InvalidToken
 from rest_framework_simplejwt.settings import api_settings
+from rest_framework_simplejwt.tokens import AccessToken
 
 
-def get_user_dict(token):
+def get_user_dict(token, force_newsletter_subscription=False):
     """Get user field values from token."""
     values = {
         field: token[token_field]
@@ -28,6 +29,12 @@ def get_user_dict(token):
     except LookupError:
         values["language"] = settings.LANGUAGE_CODE
 
+    if (
+        "has_subscribed_to_commercial_newsletter" not in values
+        and force_newsletter_subscription
+    ):
+        values["has_subscribed_to_commercial_newsletter"] = True
+
     return values
 
 
@@ -45,10 +52,16 @@ def get_user(self, validated_token):
                 _("Token contained no recognizable user identification")
             ) from exc
 
+        # force newsletter subscription if the token is from keycloak (openid)
+        force_newsletter_subscription = isinstance(validated_token, KeycloakAccessToken)
+
         def get_or_create_and_update_user():
             user, _created = self.user_model.objects.get_or_create(
                 **{api_settings.USER_ID_FIELD: user_id},
-                defaults=get_user_dict(validated_token),
+                defaults=get_user_dict(
+                    validated_token,
+                    force_newsletter_subscription=force_newsletter_subscription,
+                ),
             )
             user.update_from_token(validated_token)
             return user
@@ -94,3 +107,10 @@ class OpenApiSessionAuthenticationExtension(SessionScheme):
     target_class = (
         "joanie.core.authentication.SessionAuthenticationWithAuthenticateHeader"
     )
+
+
+class KeycloakAccessToken(AccessToken):
+    """Custom AccessToken class with `ID` token type."""
+
+    #  ruff: noqa : S105
+    token_type = "ID"

src/backend/joanie/core/models/accounts.py

@@ -118,7 +118,7 @@ def save(self, *args, **kwargs):
 
     def update_from_token(self, token):
         """Update user from token token."""
-        values = get_user_dict(token)
+        values = get_user_dict(token, force_newsletter_subscription=False)
         for key, value in values.items():
             if value != getattr(self, key):
                 User.objects.filter(

src/backend/joanie/core/utils/jwt_tokens.py

@@ -4,14 +4,17 @@
 
 from datetime import datetime, timedelta
 
+from django.conf import settings
+
 from rest_framework_simplejwt.tokens import AccessToken
 
 from joanie.core import models
+from joanie.core.authentication import KeycloakAccessToken
 
 
 def generate_jwt_token_from_user(
     user: models.User, expires_at: datetime = None
-) -> AccessToken:
+) -> AccessToken | KeycloakAccessToken:
     """
     Generate a jwt token used to authenticate a user from a user registered in
     the database
@@ -24,15 +27,45 @@ def generate_jwt_token_from_user(
         token, the jwt token generated as it should
     """
     issued_at = datetime.utcnow()
-    token = AccessToken()
-    token.payload.update(
-        {
-            "email": user.email,
-            "exp": expires_at or issued_at + timedelta(days=2),
-            "iat": issued_at,
-            "language": user.language,
-            "username": user.username,
-            "full_name": user.get_full_name(),
-        }
-    )
+    if issuer := settings.SIMPLE_JWT.get("ISSUER"):
+        token = KeycloakAccessToken()
+        token.payload.update(
+            {
+                "exp": expires_at or issued_at + timedelta(days=2),
+                "iat": issued_at,
+                "auth_time": 1768924092,
+                "jti": "c7ee46da-8127-51d1-35b1-3f07fa1a49a5",
+                "iss": issuer,
+                "aud": "keycloak-client",
+                "sub": "095009db-b774-4e26-ab58-5e55c1474d98",
+                "typ": "ID",
+                "azp": "keycloak-client",
+                "nonce": "1a21d63a-930b-457f-9445-0858645f77ba",
+                "sid": "9f00242d-9199-80d8-178a-5f9c73968385",
+                "at_hash": "SPRUypol4zCSgENoM3764g",
+                "acr": "0",
+                "s_hash": "YFa348xSzi5FBMi4x9w6jg",
+                "email_verified": False,
+                "name": user.get_full_name(),
+                "preferred_username": user.username,
+                "given_name": user.first_name,
+                "family_name": user.last_name,
+                "email": user.email,
+                "locale": user.language,
+            }
+        )
+        backend = token.get_token_backend()
+        backend.algorithm = "HS256"
+    else:
+        token = AccessToken()
+        token.payload.update(
+            {
+                "email": user.email,
+                "exp": expires_at or issued_at + timedelta(days=2),
+                "iat": issued_at,
+                "language": user.language,
+                "username": user.username,
+                "full_name": user.get_full_name(),
+            }
+        )
     return token

src/backend/joanie/settings.py

@@ -341,8 +341,17 @@ class Base(Configuration):
         "AUTH_HEADER_TYPES": ("Bearer",),
         "AUTH_HEADER_NAME": "HTTP_AUTHORIZATION",
         "USER_ID_FIELD": "username",
-        "USER_ID_CLAIM": "username",
-        "AUTH_TOKEN_CLASSES": ("rest_framework_simplejwt.tokens.AccessToken",),
+        "USER_ID_CLAIM": values.Value(
+            "username", environ_name="KEYCLOAK_USERNAME_CLAIM"
+        ),
+        "ISSUER": values.Value(None, environ_name="KEYCLOAK_ISSUER"),
+        "JWK_URL": values.Value(None, environ_name="KEYCLOAK_JWK_URL"),
+        # "TOKEN_TYPE_CLAIM": "typ",
+        "TOKEN_TYPE_CLAIM": values.Value("token_type", environ_name="JWT_TYPE_CLAIM"),
+        "AUTH_TOKEN_CLASSES": (
+            "rest_framework_simplejwt.tokens.AccessToken",
+            "joanie.core.authentication.KeycloakAccessToken",
+        ),
     }
     JWT_USER_FIELDS_SYNC = values.DictValue(
         {

src/backend/joanie/tests/base.py

@@ -13,6 +13,7 @@
 from rest_framework_simplejwt.tokens import AccessToken
 
 from joanie.core import enums
+from joanie.core.authentication import KeycloakAccessToken
 from joanie.core.models import ActivityLog
 from joanie.core.utils.jwt_tokens import generate_jwt_token_from_user
 from joanie.core.utils.sentry import serialize_data
@@ -43,16 +44,46 @@ def get_user_token(username, expires_at=None):
             token, the jwt token generated as it should
         """
         issued_at = datetime.utcnow()
-        token = AccessToken()
-        token.payload.update(
-            {
-                "email": f"{username}@funmooc.fr",
-                "exp": expires_at or issued_at + timedelta(days=2),
-                "iat": issued_at,
-                "language": settings.LANGUAGE_CODE,
-                "username": username,
-            }
-        )
+        if issuer := settings.SIMPLE_JWT.get("ISSUER"):
+            token = KeycloakAccessToken()
+            token.payload.update(
+                {
+                    "exp": expires_at or issued_at + timedelta(days=2),
+                    "iat": issued_at,
+                    "auth_time": 1768924092,
+                    "jti": "c7ee46da-8127-51d1-35b1-3f07fa1a49a5",
+                    "iss": issuer,
+                    "aud": "keycloak-client",
+                    "sub": "095009db-b774-4e26-ab58-5e55c1474d98",
+                    "typ": "ID",
+                    "azp": "keycloak-client",
+                    "nonce": "1a21d63a-930b-457f-9445-0858645f77ba",
+                    "sid": "9f00242d-9199-80d8-178a-5f9c73968385",
+                    "at_hash": "SPRUypol4zCSgENoM3764g",
+                    "acr": "0",
+                    "s_hash": "YFa348xSzi5FBMi4x9w6jg",
+                    "email_verified": False,
+                    "name": "",
+                    "preferred_username": username,
+                    "given_name": "",
+                    "family_name": "",
+                    "email": f"{username}@funmooc.fr",
+                    "locale": settings.LANGUAGE_CODE,
+                }
+            )
+            backend = token.get_token_backend()
+            backend.algorithm = "HS256"
+        else:
+            token = AccessToken()
+            token.payload.update(
+                {
+                    "email": f"{username}@funmooc.fr",
+                    "exp": expires_at or issued_at + timedelta(days=2),
+                    "iat": issued_at,
+                    "language": settings.LANGUAGE_CODE,
+                    "username": username,
+                }
+            )
         return token
 
     @staticmethod

src/backend/joanie/tests/core/test_api_base.py

@@ -4,15 +4,19 @@
 
 from datetime import datetime, timedelta
 
+from django.test import override_settings
+
 from joanie.core import factories
 from joanie.tests.base import BaseAPITestCase
 
 
 class BaseAPITestTestCase(BaseAPITestCase):
     """Test suite for BaseAPITest class"""
 
+    @override_settings(SIMPLE_JWT={"ISSUER": None})
     def test_base_api_generate_token_from_users(self):
-        """If a user is passed to the method generate_token_from_user,
+        """
+        If a user is passed to the method generate_token_from_user,
         the token attributes should correspond to the data of the user
         """
         user = factories.UserFactory(
@@ -27,3 +31,24 @@ def test_base_api_generate_token_from_users(self):
         self.assertGreater(
             token.payload.get("exp"), datetime.utcnow() + timedelta(days=1)
         )
+
+    @override_settings(SIMPLE_JWT={"ISSUER": "AnyIssuer"})
+    def test_base_api_generate_token_from_users_keycloak(self):
+        """
+        If a user is passed to the method generate_token_from_user,
+        the token attributes should correspond to the data of the user.
+        In Keycloak, preferred_username is used instead of username
+        and locale is used instead of language.
+        """
+        user = factories.UserFactory(
+            username="Sam", email="sam@fun-test.fr", language="fr-fr"
+        )
+        token = self.generate_token_from_user(user)
+        self.assertEqual("sam@fun-test.fr", token.payload.get("email"))
+        self.assertEqual("Sam", token.payload.get("preferred_username"))
+        self.assertEqual("fr-fr", token.payload.get("locale"))
+
+        # expiration date is over a day from now
+        self.assertGreater(
+            token.payload.get("exp"), datetime.utcnow() + timedelta(days=1)
+        )