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.http import HttpResponse 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"] 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["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["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 _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, }