From e4e2c15505251ccfe5e338cd9cd6e9c2b78a9058 Mon Sep 17 00:00:00 2001 From: lebaudantoine Date: Thu, 22 Jan 2026 20:15:07 +0100 Subject: [PATCH] wip refactor token generation --- src/backend/core/external_api/viewsets.py | 40 ++++---- .../core/services/jwt_token_service.py | 97 +++++++++++++++++++ 2 files changed, 114 insertions(+), 23 deletions(-) create mode 100644 src/backend/core/services/jwt_token_service.py diff --git a/src/backend/core/external_api/viewsets.py b/src/backend/core/external_api/viewsets.py index 11b6fc6b..2e02f603 100644 --- a/src/backend/core/external_api/viewsets.py +++ b/src/backend/core/external_api/viewsets.py @@ -1,6 +1,5 @@ """External API endpoints""" -from datetime import datetime, timedelta, timezone from logging import getLogger from django.conf import settings @@ -8,7 +7,6 @@ from django.contrib.auth.hashers import check_password from django.core.exceptions import SuspiciousOperation, ValidationError from django.core.validators import validate_email -import jwt from lasuite.oidc_resource_server.authentication import ResourceServerAuthentication from rest_framework import decorators, mixins, viewsets from rest_framework import ( @@ -22,6 +20,7 @@ from rest_framework import ( ) from core import api, models +from core.services.jwt_token_service import TokenService from . import authentication, permissions, serializers @@ -128,33 +127,28 @@ class ApplicationViewSet(viewsets.GenericViewSet): "Multiple user accounts share a common email." ) from e - now = datetime.now(timezone.utc) scope = " ".join(application.scopes or []) - payload = { - "iss": settings.APPLICATION_JWT_ISSUER, - "aud": settings.APPLICATION_JWT_AUDIENCE, - "iat": now, - "exp": now + timedelta(seconds=settings.APPLICATION_JWT_EXPIRATION_SECONDS), - "client_id": client_id, - "scope": scope, - "user_id": str(user.id), - "delegated": True, - } - - token = jwt.encode( - payload, - settings.APPLICATION_JWT_SECRET_KEY, + token_service = TokenService( + secret_key=settings.APPLICATION_JWT_SECRET_KEY, algorithm=settings.APPLICATION_JWT_ALG, + issuer=settings.APPLICATION_JWT_ISSUER, + audience=settings.APPLICATION_JWT_AUDIENCE, + expiration_seconds=settings.APPLICATION_JWT_EXPIRATION_SECONDS, + token_type=settings.APPLICATION_JWT_TOKEN_TYPE, + ) + + data = token_service.generate_access_token( + user, + scope, + { + "client_id": client_id, + "delegated": True, + }, ) return drf_response.Response( - { - "access_token": token, - "token_type": settings.APPLICATION_JWT_TOKEN_TYPE, - "expires_in": settings.APPLICATION_JWT_EXPIRATION_SECONDS, - "scope": scope, - }, + data, status=drf_status.HTTP_200_OK, ) diff --git a/src/backend/core/services/jwt_token_service.py b/src/backend/core/services/jwt_token_service.py new file mode 100644 index 00000000..77c3a0f1 --- /dev/null +++ b/src/backend/core/services/jwt_token_service.py @@ -0,0 +1,97 @@ +"""JWT token service.""" + +# pylint: disable=R0913,R0917 +# ruff: noqa: PLR0913 + +from datetime import datetime, timedelta, timezone +from typing import Optional + +from django.core.exceptions import ImproperlyConfigured + +import jwt + + +class TokenService: + """Generic JWT token service with configurable settings.""" + + def __init__( + self, + secret_key: str, + algorithm: str, + issuer: str, + audience: str, + expiration_seconds: int, + token_type: str, + ): + """ + Initialize the token service with custom settings. + + Args: + secret_key: Secret key for JWT encoding/decoding + algorithm: JWT algorithm (default: HS256) + issuer: Token issuer identifier + audience: Token audience identifier + expiration_seconds: Token expiration time in seconds (default: 3600) + token_type: Token type (default: Bearer) + + Raises: + ImproperlyConfigured: If secret_key is None or empty + """ + if not secret_key: + raise ImproperlyConfigured("Secret key is required.") + + self._key = secret_key + self._alg = algorithm + self._issuer = issuer + self._audience = audience + self._expiration_seconds = expiration_seconds + self._token_type = token_type + + def generate_access_token( + self, user, scope: str, extra_payload: Optional[dict] = None + ) -> dict: + """ + Generate an access token for the given user. + + Args: + user: User instance for whom to generate the token + scope: Space-separated scope string + + Returns: + Dictionary containing access_token, token_type, expires_in, and scope + """ + now = datetime.now(timezone.utc) + + payload = extra_payload.copy() if extra_payload else {} + + payload.update( + { + "iat": now, + "exp": now + timedelta(seconds=self._expiration_seconds), + "user_id": str(user.id), + } + ) + + if self._issuer: + payload["iss"] = self._issuer + if self._audience: + payload["aud"] = self._audience + if scope: + payload["scope"] = scope + + token = jwt.encode( + payload, + self._key, + algorithm=self._alg, + ) + + response = { + "access_token": token, + "token_type": self._token_type, + "expires_in": self._expiration_seconds, + } + + if scope: + response["scope"] = scope + + return response