mirror of
https://github.com/cisagov/manage.get.gov.git
synced 2025-05-15 17:17:02 +02:00
131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
# coding: utf-8
|
|
|
|
import logging
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import logout as auth_logout
|
|
from django.contrib.auth import authenticate, login
|
|
from django.http import HttpResponseRedirect
|
|
from django.shortcuts import redirect, render
|
|
from urllib.parse import parse_qs, urlencode
|
|
|
|
from djangooidc.oidc import Client
|
|
from djangooidc import exceptions as o_e
|
|
from registrar.models import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
# Initialize provider using pyOICD
|
|
OP = getattr(settings, "OIDC_ACTIVE_PROVIDER")
|
|
CLIENT = Client(OP)
|
|
logger.debug("client initialized %s" % CLIENT)
|
|
except Exception as err:
|
|
CLIENT = None # type: ignore
|
|
logger.warning(err)
|
|
logger.warning("Unable to configure OpenID Connect provider. Users cannot log in.")
|
|
|
|
|
|
def error_page(request, error):
|
|
"""Display a sensible message and log the error."""
|
|
logger.error(error)
|
|
if isinstance(error, o_e.AuthenticationFailed):
|
|
return render(
|
|
request,
|
|
"401.html",
|
|
context={
|
|
"friendly_message": error.friendly_message,
|
|
"log_identifier": error.locator,
|
|
},
|
|
status=401,
|
|
)
|
|
if isinstance(error, o_e.InternalError):
|
|
return render(
|
|
request,
|
|
"500.html",
|
|
context={
|
|
"friendly_message": error.friendly_message,
|
|
"log_identifier": error.locator,
|
|
},
|
|
status=500,
|
|
)
|
|
if isinstance(error, Exception):
|
|
return render(request, "500.html", status=500)
|
|
|
|
|
|
def openid(request):
|
|
"""Redirect the user to an authentication provider (OP)."""
|
|
request.session["next"] = request.GET.get("next", "/")
|
|
|
|
try:
|
|
return CLIENT.create_authn_request(request.session)
|
|
except Exception as err:
|
|
return error_page(request, err)
|
|
|
|
|
|
def login_callback(request):
|
|
"""Analyze the token returned by the authentication provider (OP)."""
|
|
try:
|
|
query = parse_qs(request.GET.urlencode())
|
|
userinfo = CLIENT.callback(query, request.session)
|
|
# test for need for identity verification and if it is satisfied
|
|
# if not satisfied, redirect user to login with stepped up acr_value
|
|
if requires_step_up_auth(userinfo):
|
|
# add acr_value to request.session
|
|
request.session["acr_value"] = CLIENT.get_step_up_acr_value()
|
|
return CLIENT.create_authn_request(request.session)
|
|
user = authenticate(request=request, **userinfo)
|
|
if user:
|
|
login(request, user)
|
|
logger.info("Successfully logged in user %s" % user)
|
|
return redirect(request.session.get("next", "/"))
|
|
else:
|
|
raise o_e.BannedUser()
|
|
except Exception as err:
|
|
return error_page(request, err)
|
|
|
|
|
|
def requires_step_up_auth(userinfo):
|
|
"""if User.needs_identity_verification and step_up_acr_value not in
|
|
ial returned from callback, return True"""
|
|
step_up_acr_value = CLIENT.get_step_up_acr_value()
|
|
acr_value = userinfo.get("ial", "")
|
|
uuid = userinfo.get("sub", "")
|
|
email = userinfo.get("email", "")
|
|
return User.needs_identity_verification(email, uuid) and acr_value != step_up_acr_value
|
|
|
|
|
|
def logout(request, next_page=None):
|
|
"""Redirect the user to the authentication provider (OP) logout page."""
|
|
try:
|
|
user = request.user
|
|
request_args = {
|
|
"client_id": CLIENT.client_id,
|
|
"state": request.session["state"],
|
|
}
|
|
if (
|
|
"post_logout_redirect_uris" in CLIENT.registration_response.keys()
|
|
and len(CLIENT.registration_response["post_logout_redirect_uris"]) > 0
|
|
):
|
|
request_args.update(
|
|
{"post_logout_redirect_uri": CLIENT.registration_response["post_logout_redirect_uris"][0]}
|
|
)
|
|
url = CLIENT.provider_info["end_session_endpoint"]
|
|
url += "?" + urlencode(request_args)
|
|
return HttpResponseRedirect(url)
|
|
except Exception as err:
|
|
return error_page(request, err)
|
|
finally:
|
|
# Always remove Django session stuff - even if not logged out from OP.
|
|
# Don't wait for the callback as it may never come.
|
|
auth_logout(request)
|
|
logger.info("Successfully logged out user %s" % user)
|
|
next_page = getattr(settings, "LOGOUT_REDIRECT_URL", None)
|
|
if next_page:
|
|
request.session["next"] = next_page
|
|
|
|
|
|
def logout_callback(request):
|
|
"""Simple redirection view: after logout, redirect to `next`."""
|
|
next = request.session.get("next", "/")
|
|
return redirect(next)
|