zydronium/manage.get.gov
1
0
Fork 0
mirror of https://github.com/cisagov/manage.get.gov.git synced 2025-05-15 17:17:02 +02:00
Code Issues Projects Releases Packages Wiki Activity

Fix linter errors

Browse source
This commit is contained in:
Seamus Johnston 2022-09-22 15:57:40 -05:00
parent 00f87168e5
commit f16d657c6a
No known key found for this signature in database
GPG key ID: 2F21225985069105
5 changed files with 101 additions and 97 deletions
Download patch file Download diff file Expand all files Collapse all files

19
src/djangooidc/backends.py
View file

@ -10,6 +10,7 @@ from django.utils import timezone
logger = logging.getLogger(__name__)
class OpenIdConnectBackend(ModelBackend):
"""
This backend checks a previously performed OIDC authentication.
@ -33,18 +34,12 @@ class OpenIdConnectBackend(ModelBackend):
# Some OP may actually choose to withhold some information, so we must
# test if it is present
openid_data = {"last_login": timezone.now()}
if "first_name" in kwargs.keys():
openid_data["first_name"] = kwargs["first_name"]
if "given_name" in kwargs.keys():
openid_data["first_name"] = kwargs["given_name"]
if "christian_name" in kwargs.keys():
openid_data["first_name"] = kwargs["christian_name"]
if "family_name" in kwargs.keys():
openid_data["last_name"] = kwargs["family_name"]
if "last_name" in kwargs.keys():
openid_data["last_name"] = kwargs["last_name"]
if "email" in kwargs.keys():
openid_data["email"] = kwargs["email"]
openid_data["first_name"] = kwargs.get("first_name", "")
openid_data["first_name"] = kwargs.get("given_name", "")
openid_data["first_name"] = kwargs.get("christian_name", "")
openid_data["last_name"] = kwargs.get("family_name", "")
openid_data["last_name"] = kwargs.get("last_name", "")
openid_data["email"] = kwargs.get("email", "")
# Note that this could be accomplished in one try-except clause, but
# instead we use get_or_create when creating unknown users since it has

10
src/djangooidc/exceptions.py
View file

@ -9,16 +9,12 @@ class OIDCException(Exception):
`.locator`, if used, should be a useful, unique identifier for
locating related log messages.
"""
status = status.INTERNAL_SERVER_ERROR
friendly_message = "A server error occurred."
locator = None
def __init__(
self,
friendly_message=None,
status=None,
locator=None
):
def __init__(self, friendly_message=None, status=None, locator=None):
if friendly_message is not None:
self.friendly_message = friendly_message
if status is not None:
@ -43,4 +39,4 @@ class InternalError(OIDCException):
class BannedUser(AuthenticationFailed):
friendly_message = "Your user is not valid in this application."
friendly_message = "Your user is not valid in this application."

132
src/djangooidc/oidc.py
View file

@ -4,11 +4,6 @@ from __future__ import unicode_literals
import logging
import json
try:
from builtins import unicode as str
except ImportError:
pass
from django.conf import settings
from django.http import HttpResponseRedirect
from Cryptodome.PublicKey.RSA import importKey
@ -50,7 +45,8 @@ class Client(oic.Client):
except Exception as err:
logger.error(err)
logger.error(
"Key jar preparation failed for %s", provider["srv_discovery_url"],
"Key jar preparation failed for %s",
provider["srv_discovery_url"],
)
raise o_e.InternalError()
@ -68,7 +64,8 @@ class Client(oic.Client):
except Exception as err:
logger.error(err)
logger.error(
"Client creation failed for %s", provider["srv_discovery_url"],
"Client creation failed for %s",
provider["srv_discovery_url"],
)
raise o_e.InternalError()
@ -81,7 +78,8 @@ class Client(oic.Client):
except Exception as err:
logger.error(err)
logger.error(
"Provider info discovery failed for %s", provider["srv_discovery_url"],
"Provider info discovery failed for %s",
provider["srv_discovery_url"],
)
raise o_e.InternalError()
@ -104,7 +102,7 @@ class Client(oic.Client):
"state": session["state"],
"nonce": session["nonce"],
"redirect_uri": self.registration_response["redirect_uris"][0],
"acr_values": self.behaviour.get("acr_value")
"acr_values": self.behaviour.get("acr_value"),
}
if extra_args is not None:
@ -160,7 +158,7 @@ class Client(oic.Client):
AuthorizationResponse,
unparsed_response,
sformat="dict",
keyjar=self.keyjar
keyjar=self.keyjar,
)
except Exception as err:
logger.error(err)
@ -171,7 +169,9 @@ class Client(oic.Client):
if isinstance(authn_response, ErrorResponse):
error = authn_response.get("error", "")
if error == "login_required":
logger.warning("User was not logged in (%s), trying again for %s" % (error, state))
logger.warning(
"User was not logged in (%s), trying again for %s" % (error, state)
)
return self.create_authn_request(session)
else:
logger.error("Unable to process response %s for %s" % (error, state))
@ -189,56 +189,19 @@ class Client(oic.Client):
raise o_e.AuthenticationFailed(locator=state)
if self.behaviour.get("response_type") == "code":
try:
# request a new token by which we may interact with OP
# on behalf of the user
token_response = self.do_access_token_request(
scope="openid",
state=authn_response["state"],
request_args={
"code": authn_response["code"],
"redirect_uri": self.registration_response["redirect_uris"][0],
"client_id": self.client_id,
"client_secret": self.client_secret,
},
authn_method=self.registration_response[
"token_endpoint_auth_method"
],
)
except Exception as err:
logger.error(err)
logger.error("Unable to obtain access token for %s" % state)
raise o_e.AuthenticationFailed(locator=state)
# need an access token to get user info (and to log the user out later)
self._request_token(
authn_response["state"], authn_response["code"], session
)
# ErrorResponse is not raised, it is passed back...
if isinstance(token_response, ErrorResponse):
logger.error("Unable to get token (%s) for %s" % (token_response.get("error",""), state))
raise o_e.AuthenticationFailed(locator=state)
user_info = self._get_user_info(state, session)
logger.debug("token response %s" % token_response)
try:
# get the token and other bits of info
id_token = token_response["id_token"]._dict
if id_token["nonce"] != session["nonce"]:
logger.error("Received nonce not the same as expected for %s" % state)
raise o_e.AuthenticationFailed(locator=state)
session["id_token"] = id_token
session["id_token_raw"] = getattr(self, "id_token_raw", None)
session["access_token"] = token_response["access_token"]
session["refresh_token"] = token_response.get("refresh_token", "")
session["expires_in"] = token_response.get("expires_in", "")
self.id_token[authn_response["state"]] = getattr(self, "id_token_raw", None)
except Exception as err:
logger.error(err)
logger.error("Unable to parse access token response for %s" % state)
raise o_e.AuthenticationFailed(locator=state)
return user_info
def _get_user_info(self, state, session):
"""Get information from OP about the user."""
scopes = list(self.behaviour.get("user_info_request", []))
scopes.append("openid")
try:
# get info about the user from OP
info_response = self.do_user_info_request(
@ -253,13 +216,62 @@ class Client(oic.Client):
# ErrorResponse is not raised, it is passed back...
if isinstance(info_response, ErrorResponse):
logger.error("Unable to get user info (%s) for %s" % (info_response.get("error",""), state))
logger.error(
"Unable to get user info (%s) for %s"
% (info_response.get("error", ""), state)
)
raise o_e.AuthenticationFailed(locator=state)
user_info = info_response.to_dict()
logger.debug("user info: %s" % user_info)
logger.debug("user info: %s" % info_response)
return info_response.to_dict()
return user_info
def _request_token(self, state, code, session):
"""Request a token from OP to allow us to then request user info."""
try:
token_response = self.do_access_token_request(
scope="openid",
state=state,
request_args={
"code": code,
"redirect_uri": self.registration_response["redirect_uris"][0],
"client_id": self.client_id,
"client_secret": self.client_secret,
},
authn_method=self.registration_response["token_endpoint_auth_method"],
)
except Exception as err:
logger.error(err)
logger.error("Unable to obtain access token for %s" % state)
raise o_e.AuthenticationFailed(locator=state)
# ErrorResponse is not raised, it is passed back...
if isinstance(token_response, ErrorResponse):
logger.error(
"Unable to get token (%s) for %s"
% (token_response.get("error", ""), state)
)
raise o_e.AuthenticationFailed(locator=state)
logger.debug("token response %s" % token_response)
try:
# get the token and other bits of info
id_token = token_response["id_token"]._dict
if id_token["nonce"] != session["nonce"]:
logger.error("Received nonce not the same as expected for %s" % state)
raise o_e.AuthenticationFailed(locator=state)
session["id_token"] = id_token
session["id_token_raw"] = getattr(self, "id_token_raw", None)
session["access_token"] = token_response["access_token"]
session["refresh_token"] = token_response.get("refresh_token", "")
session["expires_in"] = token_response.get("expires_in", "")
self.id_token[state] = getattr(self, "id_token_raw", None)
except Exception as err:
logger.error(err)
logger.error("Unable to parse access token response for %s" % state)
raise o_e.AuthenticationFailed(locator=state)
def store_response(self, resp, info):
"""Make raw ID token available for internal use."""

32
src/djangooidc/views.py
View file

@ -2,18 +2,13 @@
import logging
try:
from urllib.parse import parse_qs, urlencode
except ImportError:
from urllib import urlencode
from urlparse import parse_qs
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
@ -35,9 +30,9 @@ def error_page(request, error):
"401.html",
context={
"friendly_message": error.friendly_message,
"log_identifier": error.locator
"log_identifier": error.locator,
},
status=401
status=401,
)
if isinstance(error, o_e.InternalError):
return render(
@ -45,13 +40,14 @@ def error_page(request, error):
"500.html",
context={
"friendly_message": error.friendly_message,
"log_identifier": error.locator
"log_identifier": error.locator,
},
status=500
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", "/")
@ -61,6 +57,7 @@ def openid(request):
except Exception as err:
return error_page(request, err)
def login_callback(request):
"""Analyze the token returned by the authentication provider (OP)."""
try:
@ -90,11 +87,13 @@ def logout(request, next_page=None):
"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]
})
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)
@ -110,6 +109,7 @@ def logout(request, next_page=None):
if next_page:
request.session["next"] = next_page
def logout_callback(request):
"""Simple redirection view: after logout, redirect to `next`."""
next = request.session["next"] if "next" in request.session.keys() else "/"