55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
import structlog as structlog
|
|
from authlib.integrations.base_client import OAuthError
|
|
from django.conf import settings
|
|
from django.shortcuts import redirect
|
|
from django.http import HttpResponse
|
|
from sentry_sdk import capture_exception
|
|
from django.contrib.auth import login as dj_login, get_user_model
|
|
|
|
from vbv_lernwelt.sso.client import oauth
|
|
from vbv_lernwelt.sso.jwt import decode_jwt
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
OAUTH_SUCCESS_REDIRECT = 'profile'
|
|
OAUTH_FAIL_REDIRECT = 'login-error'
|
|
|
|
|
|
def login(request):
|
|
oauth_client = oauth.create_client(settings.OAUTH["client_name"])
|
|
redirect_uri = settings.OAUTH["local_redirect_uri"]
|
|
return oauth_client.authorize_redirect(request, redirect_uri)
|
|
|
|
|
|
def authorize(request):
|
|
try:
|
|
logger.debug(request)
|
|
token = getattr(oauth, settings.OAUTH["client_name"]).authorize_access_token(request)
|
|
deocded_token = decode_jwt(token["access_token"])
|
|
except OAuthError as e:
|
|
logger.error(f'OAuth error: {e}')
|
|
if not settings.DEBUG:
|
|
capture_exception(e)
|
|
return redirect(f'/{OAUTH_FAIL_REDIRECT}?state=someerror') # to be defined
|
|
|
|
user_data = _user_data_from_token_data(deocded_token)
|
|
user, created = get_user_model().objects.create_or_update_by_email(user_data["email"], user_data["first_name"],
|
|
user_data["last_name"], user_data["username"])
|
|
|
|
dj_login(request, user)
|
|
# todo: redirect to other page if new user
|
|
return redirect(f'/{OAUTH_SUCCESS_REDIRECT}?state=success') # to be defined
|
|
|
|
|
|
def foo(request):
|
|
return HttpResponse(settings.OAUTH["server_metadata_url"])
|
|
|
|
|
|
def _user_data_from_token_data(token: dict) -> dict:
|
|
return {
|
|
"first_name": token.get("given_name", ""),
|
|
"last_name": token.get("family_name", ""),
|
|
"username": token.get("preferred_username", ""),
|
|
"email": token.get("email", ""),
|
|
}
|