# OAuth2 update

* integrates authlib into project * adds authlib package to requirements.txt * refactors internal konova/utils/generators/generate_token method to use authlib's generate_token method and act as wrapper for backwards compatibility * refactors OAuth views to be based on authlib's oauth client implementation * refactors oauth model to work with authlib's oauth2 tokens
2026-10-06 11:27:50 +02:00
parent 8de467ec24
commit e3d5ee82ee
4 changed files with 77 additions and 90 deletions
@@ -1,7 +1,9 @@
+import datetime
 import json
 from datetime import timedelta

 import requests
+from authlib.oauth2.rfc6749 import OAuth2Token
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import models
 from django.utils import timezone
@@ -78,27 +80,25 @@ class OAuthToken(UuidModel):
        return str(self.access_token)

    @staticmethod
-    def from_access_token_response(access_token_data: str, received_on):
+    def create_from_oauth2_token(oauth2_token: OAuth2Token):
        """
        Creates an OAuthToken based on retrieved access token data (OAuth2.0 specification)

        Args:
-            access_token_data (str): OAuth2.0 response data
-            received_on (): Timestamp when the response has been received
+            oauth2_token (str): OAuth2.0 response data

        Returns:

        """
        oauth_token = OAuthToken()
-        data = json.loads(access_token_data)

-        oauth_token.access_token = data.get("access_token")
-        oauth_token.refresh_token = data.get("refresh_token")
+        oauth_token.access_token = oauth2_token.get("access_token")
+        oauth_token.refresh_token = oauth2_token.get("refresh_token")
+        oauth_token.expires_on = datetime.datetime.fromtimestamp(oauth2_token.get("expires_at"))

-        expires_on = received_on + timedelta(
-            seconds=(data.get("expires_in") + OAuthToken.ASSUMED_LATENCY)
-        )
-        oauth_token.expires_on = expires_on
+        assert (oauth_token.access_token is not None and
+                oauth_token.refresh_token is not None and
+                oauth_token.expires_on is not None)

        return oauth_token

@@ -8,6 +8,9 @@ Created on: 09.11.20
 import secrets
 import string

+from authlib.common.security import generate_token as authlib_generate_token
+
+

 def generate_token(length: int = 64) -> str:
    """ Shortcut for default generating of e.g. API token
@@ -15,11 +18,7 @@ def generate_token(length: int = 64) -> str:
    Returns:
        token (str)
    """
-    return generate_random_string(
-        length=length,
-        use_numbers=True,
-        use_letters_lc=True
-    )
+    return authlib_generate_token(length)


 def generate_random_string(length: int, use_numbers: bool = False, use_letters_lc: bool = False, use_letters_uc: bool = False) -> str:
@@ -7,33 +7,52 @@ Created on: 26.04.24
 """
 import base64
 import hashlib
-from urllib.parse import urlencode

-import requests
+from authlib.common.security import generate_token
+from authlib.integrations.base_client import OAuthError
+from authlib.integrations.django_client import OAuth
+
 from django.contrib.auth import login
-from django.http import HttpRequest
+from django.http import HttpRequest, HttpResponse
 from django.shortcuts import redirect
 from django.urls import reverse
-from django.utils.timezone import now
 from django.views import View

 from api.models import OAuthToken
 from konova.sub_settings.django_settings import BASE_URL
-from konova.sub_settings.sso_settings import SSO_SERVER_BASE, OAUTH_CODE_VERIFIER, OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET
+from konova.sub_settings.sso_settings import SSO_SERVER_BASE, OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET


+
+_OAUTH = OAuth()
+_OAUTH.register(
+    name="sso",
+    client_id=OAUTH_CLIENT_ID,
+    client_secret=OAUTH_CLIENT_SECRET,
+    authorize_url=f"{SSO_SERVER_BASE}o/authorize/",
+    access_token_url=f"{SSO_SERVER_BASE}o/token/",
+    code_challenge_method="S256",
+    client_kwargs={},
+)
+
 class OAuthLoginView(View):
-    """
-    Starts OAuth Login procedure
-        -> AnonymousUser is redirected to SSO component using specific parameters
-        -> After successful login (in SSO component), user will be redirected to a specific callback url (OAuthCallbackView)
-        -> Callback view uses retrieved authorization token to get a proper access token from SSO component
-        -> SSO component answers with access token
-        -> OAuthCallbackView uses token in Authorization header to access user data of logged-in user in SSO component
-        -> OAuthCallbackView creates/updates user
-        -> OAuthCallbackView logs in user and redirects to default home view
+    """ View for OAuth2 login step

    """
+    def get(self, request: HttpRequest, *args, **kwargs):
+        redirect_uri = f'{BASE_URL}{reverse("oauth-callback")}'
+
+        # create new PKCE-Verifier
+        code_verifier, code_challenge = self.__create_code_challenge()
+
+        # keep verifier in session, so it can be read out for callback
+        request.session["oauth_code_verifier"] = code_verifier
+
+        return _OAUTH.sso.authorize_redirect(
+            request,
+            redirect_uri,
+            code_verifier=code_verifier,
+        )

    def __create_code_challenge(self):
        """
@@ -44,82 +63,49 @@ class OAuthLoginView(View):
        Returns:

        """
-        code_verifier = OAUTH_CODE_VERIFIER
+        code_verifier = generate_token(128)

        code_challenge = hashlib.sha256(code_verifier.encode('utf-8')).digest()
        code_challenge = base64.urlsafe_b64encode(code_challenge).decode('utf-8').replace('=', '')
        return code_verifier, code_challenge

-    def get(self, request: HttpRequest, *args, **kwargs):
-        """
-        Redirects user to OAuth SSO webservice for credential based login there
-
-        Args:
-            request ():
-            *args ():
-            **kwargs ():
-
-        Returns:
-
-        """
-        oauth_authentication_code_url = f"{SSO_SERVER_BASE}o/authorize/"
-        redirect_uri = f'{BASE_URL}{reverse("oauth-callback")}'
-
-        code_verifier, code_challenge = self.__create_code_challenge()
-
-        urlencode_params = urlencode(
-            {
-                "response_type": "code",
-                "code_challenge": code_challenge,
-                "code_challenge_method": "S256",
-                "client_id": OAUTH_CLIENT_ID,
-                "redirect_uri": redirect_uri,
-            }
-        )
-        url = f"{oauth_authentication_code_url}?{urlencode_params}"
-        return redirect(url)
-
-
 class OAuthCallbackView(View):
-    """
-    Callback view for OAuth2.0 authentication token.
-    Authentication tokens will be exchanged for access token.
-    Access Token will be used for fetching user data from SSO component.
-    User data will be used for creating/updating user data inside this app.
-    User will be logged-in and redirected to default home view.
+    """ View for OAuth2 callback step

    """
-
    def get(self, request: HttpRequest, *args, **kwargs):
-        authentication_code = request.GET.get("code")
-        oauth_acces_token_url = f"{SSO_SERVER_BASE}o/token/"
-
-        callback_url = f'{BASE_URL}{reverse("oauth-callback")}'
-
-        params = {
-            "grant_type": "authorization_code",
-            "code": authentication_code,
-            "redirect_uri": callback_url,
-            "code_verifier": OAUTH_CODE_VERIFIER,
-            "client_id": OAUTH_CLIENT_ID,
-            "client_secret": OAUTH_CLIENT_SECRET
-        }
-        access_code_response = requests.post(
-            oauth_acces_token_url,
-            data=params
+        # Get code verifier (created on oatuh-login step) from session
+        code_verifier = request.session.pop(
+            "oauth_code_verifier",
+            None,
        )
-        received_on = now()
+        if not code_verifier:
+            return HttpResponse(
+                "PKCE code_verifier missing or is invalid.",
+                status=400,
+            )

-        access_code_response_body = access_code_response.content.decode("utf-8")
-        status_code_invalid = access_code_response.status_code != 200
-        if status_code_invalid:
-            raise RuntimeError(f"OAuth access token could not be fetched: {access_code_response.text}")
+        try:
+            token = _OAUTH.sso.authorize_access_token(
+                request
+            )
+        except OAuthError as e:
+            print(e)
+            return HttpResponse(
+                "OAuth-Login failed while authorizing with OAuth.",
+                status=400,
+            )
+        try:
+            oauth_token = OAuthToken.create_from_oauth2_token(token)
+        except AssertionError:
+            return HttpResponse(
+                "OAuth-Login failed due to unexpected token content.",
+                status=400,
+            )

-        oauth_token = OAuthToken.from_access_token_response(access_code_response_body, received_on)
        oauth_token.save()
        user = oauth_token.update_and_get_user()
        user.oauth_replace_token(oauth_token)

        login(request, user)
        return redirect("home")
-
@@ -1,6 +1,7 @@
 amqp==5.3.1
 asgiref==3.12.1
 async-timeout==5.0.1
+Authlib==1.8.0
 beautifulsoup4==4.15.0
 billiard==4.2.4
 cached-property==2.0.1
@@ -30,6 +31,7 @@ gunicorn==26.2.0
 idna==3.19
 importlib_metadata==9.0.1
 itsdangerous==2.2.0
+joserfc==1.7.5
 jwcrypto==1.6.0
 kombu==5.6.2
 oauthlib==3.3.1