gevorkyan.aa/edx-oauth2-wordpress-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gets if response is not a dict. catch User not found exception

Browse Source
This commit is contained in:
lpm0073 2022-10-04 17:40:51 -05:00
parent 953bab77cb
commit 4147adf7d0
1 changed files with 42 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
wp_oauth_backend/wp_oauth.py
View File

@ -135,38 +135,43 @@ class StepwiseMathWPOAuth2(BaseOAuth2):
tainted = False tainted = False
if not response: if not response:
logger.warning('get_user_details() - response object is missing.') logger.warning('get_user_details() - response object is missing.')
tainted = True tainted = True
if type(response)==dict: if type(response)!=dict:
# a def in the third_party_auth pipeline list calls get_user_details() after its already logger.warning('get_user_details() - was expecting a response object of type dict but received an object of type {t}'.format(
# been called once. i don't know why. but, it passes the original get_user_details() dict t=type(response)
# enhanced with additional token-related keys. if we receive this modified dict then we ))
# should pass it along to the next defs in the pipeline. tainted = True
#
# If most of the original keys (see dict definition below) exist in the response object
# then we can assume that this is our case.
qc_keys = ['id', 'date_joined', 'email', 'first_name', 'fullname', 'is_staff', 'is_superuser', 'last_name', 'username']
if all(key in response for key in qc_keys):
if VERBOSE_LOGGING:
logger.info('get_user_details() - detected an enhanced get_user_details() dict in the response: {response}'.format(
response=json.dumps(response, sort_keys=True, indent=4)
))
return response
# otherwise we pobably received the default response from the oauth provider based on # a def in the third_party_auth pipeline list calls get_user_details() after its already
# the scopes 'basic' 'email' 'profile'. We'll check a few of the most important keys to see # been called once. i don't know why. but, it passes the original get_user_details() dict
# if they exist. # enhanced with additional token-related keys. if we receive this modified dict then we
if ('ID' not in response.keys()) or ('user_email' not in response.keys()) or ('user_login' not in response.keys()): # should pass it along to the next defs in the pipeline.
logger.warning('get_user_details() - response object is missing one or more required keys: {response}'.format( #
# If most of the original keys (see dict definition below) exist in the response object
# then we can assume that this is our case.
qc_keys = ['id', 'date_joined', 'email', 'first_name', 'fullname', 'is_staff', 'is_superuser', 'last_name', 'username']
if all(key in response for key in qc_keys):
if VERBOSE_LOGGING:
logger.info('get_user_details() - detected an enhanced get_user_details() dict in the response: {response}'.format(
response=json.dumps(response, sort_keys=True, indent=4) response=json.dumps(response, sort_keys=True, indent=4)
)) ))
tainted = True return response
else:
if VERBOSE_LOGGING: # otherwise we pobably received the default response from the oauth provider based on
logger.info('get_user_details() - start. response: {response}'.format( # the scopes 'basic' 'email' 'profile'. We'll check a few of the most important keys to see
response=json.dumps(response, sort_keys=True, indent=4) # if they exist.
)) if ('ID' not in response.keys()) or ('user_email' not in response.keys()) or ('user_login' not in response.keys()):
logger.warning('get_user_details() - response object is missing one or more required keys: {response}'.format(
response=json.dumps(response, sort_keys=True, indent=4)
))
tainted = True
else:
if VERBOSE_LOGGING:
logger.info('get_user_details() - start. response: {response}'.format(
response=json.dumps(response, sort_keys=True, indent=4)
))
if tainted and self._user_details: if tainted and self._user_details:
logger.warning('get_user_details() - returning cached results. user_details: {user_details}'.format( logger.warning('get_user_details() - returning cached results. user_details: {user_details}'.format(
@ -178,6 +183,9 @@ class StepwiseMathWPOAuth2(BaseOAuth2):
logger.error('response object is missing or misformed, and no cached results were found. Cannot get user details from oauth provider.') logger.error('response object is missing or misformed, and no cached results were found. Cannot get user details from oauth provider.')
return None return None
# ---------------------------------------------------------------------
# build and internally cache the get_user_details() dict
# ---------------------------------------------------------------------
# try to parse out the first and last names # try to parse out the first and last names
split_name = response.get('display_name', '').split() split_name = response.get('display_name', '').split()
@ -189,7 +197,6 @@ class StepwiseMathWPOAuth2(BaseOAuth2):
super_user = 'administrator' in user_roles super_user = 'administrator' in user_roles
is_staff = 'administrator' in user_roles is_staff = 'administrator' in user_roles
# build the get_user_details() dict
self._user_details = { self._user_details = {
'id': int(response.get('ID'), 0), 'id': int(response.get('ID'), 0),
'username': response.get('user_login', ''), 'username': response.get('user_login', ''),
@ -239,10 +246,12 @@ class StepwiseMathWPOAuth2(BaseOAuth2):
# add syncronization of any data fields that get missed by the built-in # add syncronization of any data fields that get missed by the built-in
# open edx third party authentication sync functionality. # open edx third party authentication sync functionality.
user=User.objects.get(username=user_details['username']) try:
# this gets called just prior to account creation for
if not user: # new users, hence, we need to catch DoesNotExist
# this seems exceedingly unlikely, but, you never know. # exceptions.
user=User.objects.get(username=user_details['username'])
except User.DoesNotExist:
return user_details return user_details
if (user.is_superuser != user_details['is_superuser']) or (user.is_staff != user_details['is_staff']): if (user.is_superuser != user_details['is_superuser']) or (user.is_staff != user_details['is_staff']):