GSGGR-156 implement authentication for the frontend-authenticated user

.env.sample

@@ -59,3 +59,4 @@ OIDC_OP_BASE_URL="please set oidc op base url"
 ZITADEL_PROJECT="set_zitadel_project"
 OIDC_RP_CLIENT_ID="set oidc rp client id"
 OIDC_RP_CLIENT_SECRET="set oidc rp client secret"
+OIDC_PRIVATE_KEYFILE="keyfile"

Dockerfile

@@ -22,3 +22,4 @@ RUN mv -vn /app/geoshop_back/default_settings.py /app/geoshop_back/settings.py
 RUN mv .env.sample .env
 RUN python3 manage.py compilemessages --locale=fr
 RUN rm -f .env
+

default_settings.py

@@ -299,6 +299,7 @@ def discover_endpoints(discovery_url: str) -> dict:
         "token_endpoint": provider_config["token_endpoint"],
         "userinfo_endpoint": provider_config["userinfo_endpoint"],
         "jwks_uri": provider_config["jwks_uri"],
+        "introspection_endpoint": provider_config["introspection_endpoint"],
     }
 
 AUTHENTICATION_BACKENDS = (
@@ -326,6 +327,8 @@ def discover_endpoints(discovery_url: str) -> dict:
     OIDC_OP_TOKEN_ENDPOINT = discovery_info["token_endpoint"]
     OIDC_OP_USER_ENDPOINT = discovery_info["userinfo_endpoint"]
     OIDC_OP_JWKS_ENDPOINT = discovery_info["jwks_uri"]
+    OIDC_OP_AUTHORIZATION_ENDPOINT = discovery_info["authorization_endpoint"]
+    OIDC_PRIVATE_KEYFILE = os.environ.get("OIDC_PRIVATE_KEYFILE")
 
     LOGIN_REDIRECT_URL = os.environ.get("OIDC_REDIRECT_BASE_URL") + "/oidc/callback"
     LOGOUT_REDIRECT_URL = os.environ.get("OIDC_REDIRECT_BASE_URL") + "/"

oidc.py

@@ -1,26 +1,112 @@
+import json
+import requests
+import time
+
+from django.views.generic import View
+from rest_framework_simplejwt.tokens import RefreshToken
 from mozilla_django_oidc.auth import OIDCAuthenticationBackend
+from authlib.jose import jwt
 from django.conf import settings
-from django.shortcuts import render
+from django.utils.decorators import method_decorator
+from django.views.decorators.csrf import csrf_exempt
+from django.http import JsonResponse
+from django.contrib.auth import get_user_model
+
+UserModel = get_user_model()
 
 def status(request):
-    return {
-        "OIDC_ENABLED": settings.OIDC_ENABLED,
-    }
+    return {"OIDC_ENABLED": settings.OIDC_ENABLED}
+
 
-def updateUser(user, claims):
+def _updateUser(user, claims):
     user.email = claims.get("email")
     user.first_name = claims.get("given_name")
     user.last_name = claims.get("family_name")
     return user
 
+
+def _read_private_key(keyfile):
+    with open(keyfile, "r") as f:
+        data = json.load(f)
+        return {
+            "client_id": data["clientId"],
+            "key_id": data["keyId"],
+            "private_key": data["key"],
+        }
+
+
+class FrontendAuthentication(View):
+
+    def __init__(self):
+        super().__init__()
+        self.private_key = _read_private_key(settings.OIDC_PRIVATE_KEYFILE)
+
+    def _get_jwt_token(self):
+        return jwt.encode(
+            {"alg": "RS256", "kid": self.private_key["key_id"]},
+            {
+                "iss": self.private_key["client_id"],
+                "sub": self.private_key["client_id"],
+                "aud": settings.OIDC_OP_BASE_URL,
+                "exp": int(time.time() + 3600),
+                "iat": int(time.time()),
+            },
+            self.private_key["private_key"])
+
+    def _resolve_user_data(self, token: str):
+        resp = requests.post(
+            settings.OIDC_INTROSPECT_URL,
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+            data={
+                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
+                "client_assertion": self._get_jwt_token(),
+                "token": token,
+            },
+        )
+        resp.raise_for_status()
+        return json.loads(resp.content)
+
+    @method_decorator(csrf_exempt)
+    def dispatch(self, request, *args, **kwargs):
+        return super().dispatch(request, *args, **kwargs)
+
+    def post(self, request):
+        # Handle JSON error
+        # TODO: Test missing id token
+        # TODO: Localize error response
+        # TODO: Test same token multiple times
+        # TODO: Test user does not exist
+        # TODO: Test user exists, but unauthrized
+        # TODO: Test user exists and authorized
+        token_data = json.loads(request.body.decode("utf-8"))
+        if "token" not in token_data:
+            return JsonResponse({"error": "No token provided"}, status=400)
+
+        user_data = self._resolve_user_data(token_data["token"])
+        try:
+            user = UserModel.objects.get(username=user_data["email"])
+        except UserModel.DoesNotExist:
+            user = UserModel.objects.create_user(username=user_data["email"])
+        _updateUser(user, user_data)
+        user.save()
+
+        token = RefreshToken.for_user(user)
+        return JsonResponse({"access": str(token.access_token), "refresh": str(token)})
+
+
 class PermissionBackend(OIDCAuthenticationBackend):
 
+    def authenticate_header(self, request):
+        # TODO: Test if header exists
+        token = request.headers["Authorization"].replace("Bearer ", "")
+        return token
+
     def create_user(self, claims):
         user = self.UserModel.objects.create_user(username=claims.get("email"))
-        updateUser(user, claims)
+        _updateUser(user, claims)
         user.save()
         return user
 
     def update_user(self, user, claims):
-        updateUser(user, claims).save()
+        _updateUser(user, claims).save()
         return user