54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
import structlog as structlog
|
|
from authlib.integrations.base_client import OAuthError
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model, login as dj_login
|
|
from django.shortcuts import redirect
|
|
from sentry_sdk import capture_exception
|
|
|
|
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"]
|
|
language = request.GET.get("lang", "de")
|
|
return oauth_client.authorize_redirect(request, redirect_uri, lang=language)
|
|
|
|
|
|
def authorize(request):
|
|
try:
|
|
logger.debug(request)
|
|
token = getattr(oauth, settings.OAUTH["client_name"]).authorize_access_token(
|
|
request
|
|
)
|
|
deocded_token = decode_jwt(token["id_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)
|
|
|
|
dj_login(request, user)
|
|
# todo: redirect to other page if new user
|
|
return redirect(f"/{OAUTH_SUCCESS_REDIRECT}?state=success") # to be defined
|
|
|
|
|
|
def _user_data_from_token_data(token: dict) -> dict:
|
|
first_email = token.get("emails", [""])[0]
|
|
return {
|
|
"first_name": token.get("given_name", ""),
|
|
"last_name": token.get("family_name", ""),
|
|
"username": token.get("preferred_username", first_email),
|
|
"email": first_email,
|
|
"oid": token.get("oid"),
|
|
}
|