395_OAuth2_refactoring
#396
Merged
mpeltriaux
merged 5 commits from 395_OAuth2_refactoring into master 5 months ago
@ -0,0 +1,26 @@
|
||||
# Generated by Django 5.0.4 on 2024-04-30 07:20
|
||||
|
||||
import uuid
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('api', '0002_alter_apiusertoken_valid_until'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='OAuthToken',
|
||||
fields=[
|
||||
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
|
||||
('access_token', models.CharField(db_comment='OAuth access token', max_length=255)),
|
||||
('refresh_token', models.CharField(db_comment='OAuth refresh token', max_length=255)),
|
||||
('expires_on', models.DateTimeField(db_comment='When the token will be expired')),
|
||||
],
|
||||
options={
|
||||
'abstract': False,
|
||||
},
|
||||
),
|
||||
]
|
||||
@ -0,0 +1,130 @@
|
||||
"""
|
||||
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
|
||||
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.utils.timezone import now
|
||||
from django.views import View
|
||||
|
||||
from api.models import OAuthToken
|
||||
from konova.sub_settings.sso_settings import SSO_SERVER_BASE, OAUTH_CODE_VERIFIER, OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET
|
||||
|
||||
|
||||
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
|
||||
|
||||
"""
|
||||
|
||||
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 for credential based login there
|
||||
|
||||
Args:
|
||||
request ():
|
||||
*args ():
|
||||
**kwargs ():
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
oauth_authentication_code_url = f"{SSO_SERVER_BASE}o/authorize/"
|
||||
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": request.build_absolute_uri(
|
||||
reverse(
|
||||
"oauth-callback"
|
||||
)
|
||||
),
|
||||
}
|
||||
)
|
||||
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.
|
||||
|
||||
"""
|
||||
|
||||
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 = request.build_absolute_uri(
|
||||
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
|
||||
)
|
||||
received_on = now()
|
||||
|
||||
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}")
|
||||
|
||||
oauth_access_token = OAuthToken.from_access_token_response(access_code_response_body, received_on)
|
||||
oauth_access_token.save()
|
||||
user = oauth_access_token.update_and_get_user()
|
||||
user.oauth_replace_token(oauth_access_token)
|
||||
|
||||
login(request, user)
|
||||
return redirect("home")
|
||||
|
||||
@ -0,0 +1,20 @@
|
||||
# Generated by Django 5.0.4 on 2024-04-30 07:20
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('api', '0003_oauthtoken'),
|
||||
('user', '0008_alter_user_id'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='user',
|
||||
name='oauth_token',
|
||||
field=models.ForeignKey(blank=True, db_comment='OAuth token for the user', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to='api.oauthtoken'),
|
||||
),
|
||||
]
|
||||
@ -0,0 +1,7 @@
|
||||
"""
|
||||
Author: Michel Peltriaux
|
||||
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
|
||||
Contact: ksp-servicestelle@sgdnord.rlp.de
|
||||
Created on: 10.05.24
|
||||
|
||||
"""
|
||||
@ -0,0 +1,69 @@
|
||||
"""
|
||||
Author: Michel Peltriaux
|
||||
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
|
||||
Contact: ksp-servicestelle@sgdnord.rlp.de
|
||||
Created on: 10.05.24
|
||||
|
||||
"""
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
|
||||
from cryptography.fernet import Fernet
|
||||
from django.core.exceptions import ObjectDoesNotExist
|
||||
from django.http import HttpRequest, JsonResponse
|
||||
from django.utils.decorators import method_decorator
|
||||
from django.views import View
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
|
||||
from konova.sub_settings.sso_settings import OAUTH_CLIENT_ID
|
||||
from user.models import User
|
||||
|
||||
|
||||
class PropagateUserView(View):
|
||||
""" Receives user data to be stored in db
|
||||
|
||||
Used if new user gets access to application and needs to be created in application before first login (e.g.
|
||||
proper rights management)
|
||||
|
||||
"""
|
||||
|
||||
@method_decorator(csrf_exempt)
|
||||
def dispatch(self, request, *args, **kwargs):
|
||||
return super().dispatch(request, *args, **kwargs)
|
||||
|
||||
def post(self, request: HttpRequest, *args, **kwargs):
|
||||
# Decrypt
|
||||
encrypted_body = request.body
|
||||
hash = hashlib.md5()
|
||||
hash.update(OAUTH_CLIENT_ID.encode("utf-8"))
|
||||
key = base64.urlsafe_b64encode(hash.hexdigest().encode("utf-8"))
|
||||
fernet = Fernet(key)
|
||||
body = fernet.decrypt(encrypted_body).decode("utf-8")
|
||||
body = json.loads(body)
|
||||
|
||||
try:
|
||||
status = "updated"
|
||||
user = User.objects.get(username=body.get('username'))
|
||||
# Update user data, excluding some changes
|
||||
skipable_attrs = {
|
||||
"username",
|
||||
"is_staff",
|
||||
"is_superuser",
|
||||
}
|
||||
for _attr, _val in body.items():
|
||||
if _attr in skipable_attrs:
|
||||
continue
|
||||
setattr(user, _attr, _val)
|
||||
except ObjectDoesNotExist:
|
||||
user = User(**body)
|
||||
status = "created"
|
||||
user.set_unusable_password()
|
||||
user.save()
|
||||
|
||||
data = {
|
||||
"success": True,
|
||||
"status": status
|
||||
}
|
||||
|
||||
return JsonResponse(data)
|
||||
Loading…
Reference in New Issue