# WIP: OAuth draft implementation
* first working client implementation of oauth workflow for logging in userspull/396/head
parent
9d4a9bd122
commit
b00dd5bcc8
@ -0,0 +1,117 @@
|
||||
"""
|
||||
Author: Michel Peltriaux
|
||||
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
|
||||
Contact: ksp-servicestelle@sgdnord.rlp.de
|
||||
Created on: 26.04.24
|
||||
|
||||
"""
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import requests
|
||||
from django.contrib.auth import login
|
||||
from django.http import HttpRequest
|
||||
from django.shortcuts import redirect
|
||||
from django.urls import reverse
|
||||
from django.views import View
|
||||
|
||||
from konova.sub_settings.sso_settings import SSO_SERVER_BASE, OAUTH_CODE_VERIFIER
|
||||
from user.models import User
|
||||
|
||||
OAUTH_CLIENT_ID = "CHANGE_ME"
|
||||
OAUTH_CLIENT_SECRET = "CHANGE_ME"
|
||||
|
||||
|
||||
class OAuthCallbackView(View):
|
||||
"""
|
||||
Callback view for a OAuth2.0 authentication token.
|
||||
Authentication tokens need to be exchanged for the access token.
|
||||
"""
|
||||
|
||||
def get(self, request: HttpRequest, *args, **kwargs):
|
||||
authentication_code = request.GET.get("code")
|
||||
oauth_acces_token_url = f"{SSO_SERVER_BASE}o/token/"
|
||||
|
||||
next_callback_url = request.build_absolute_uri(
|
||||
reverse(
|
||||
"oauth-callback"
|
||||
)
|
||||
)
|
||||
|
||||
params = {
|
||||
"grant_type": "authorization_code",
|
||||
"code": authentication_code,
|
||||
"redirect_uri": next_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
|
||||
)
|
||||
|
||||
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}")
|
||||
|
||||
access_code_response_body = json.loads(access_code_response_body)
|
||||
access_token = access_code_response_body.get("access_token")
|
||||
if not access_token:
|
||||
raise RuntimeError(f"Access token response contained no token: {access_code_response_body}")
|
||||
|
||||
user = User.oauth_get_user(access_token)
|
||||
login(request, user)
|
||||
return redirect("home")
|
||||
|
||||
|
||||
class OAuthLoginView(View):
|
||||
|
||||
def __create_code_challenge(self):
|
||||
"""
|
||||
Creates a code verifier and code challenge for extra security.
|
||||
See https://django-oauth-toolkit.readthedocs.io/en/latest/getting_started.html#authorization-code for further
|
||||
information
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
code_verifier = OAUTH_CODE_VERIFIER
|
||||
|
||||
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
|
||||
|
||||
Args:
|
||||
request ():
|
||||
*args ():
|
||||
**kwargs ():
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
oauth_authentication_code_url = f"{SSO_SERVER_BASE}o/authorize/"
|
||||
code_verifier, code_challenge = self.__create_code_challenge()
|
||||
print(code_verifier)
|
||||
|
||||
urlencode_params = urlencode(
|
||||
{
|
||||
"response_type": "code",
|
||||
"code_challenge": code_challenge,
|
||||
"code_challenge_method": "S256",
|
||||
"client_id": OAUTH_CLIENT_ID,
|
||||
"redirect_uri": request.build_absolute_uri(
|
||||
reverse(
|
||||
"oauth-callback"
|
||||
)
|
||||
),
|
||||
}
|
||||
)
|
||||
url = f"{oauth_authentication_code_url}?{urlencode_params}"
|
||||
return redirect(url)
|
Loading…
Reference in New Issue