Pass scope to OIDC_IDTOKEN_PROCESSING_HOOK

The ID token processing hook might want to add claims to the ID token
conditionally based on the scope parameter.  Therefore it would be very
useful to provide the scope parameter to the processing hook.
This commit is contained in:
Tuomas Suutari 2017-07-07 22:55:18 +03:00
parent 6199a9a17e
commit eb682f23ff
4 changed files with 41 additions and 9 deletions

View file

@ -114,7 +114,7 @@ def default_after_end_session_hook(request, id_token=None, post_logout_redirect_
return None return None
def default_idtoken_processing_hook(id_token, user): def default_idtoken_processing_hook(id_token, user, scope=None):
""" """
Hook to perform some additional actions ti `id_token` dictionary just before serialization. Hook to perform some additional actions ti `id_token` dictionary just before serialization.
@ -124,6 +124,9 @@ def default_idtoken_processing_hook(id_token, user):
:param user: user for whom id_token is generated :param user: user for whom id_token is generated
:type user: User :type user: User
:param scope: scope for the token
:type scope: list[str]|None
:return: custom modified dictionary of values for `id_token` :return: custom modified dictionary of values for `id_token`
:rtype dict :rtype dict
""" """

View file

@ -53,13 +53,14 @@ def create_id_token(user, aud, nonce='', at_hash='', request=None, scope=[]):
if ('email' in scope) and getattr(user, 'email', None): if ('email' in scope) and getattr(user, 'email', None):
dic['email'] = user.email dic['email'] = user.email
processing_hook = settings.get('OIDC_IDTOKEN_PROCESSING_HOOK') processing_hooks = settings.get('OIDC_IDTOKEN_PROCESSING_HOOK')
if isinstance(processing_hook, (list, tuple)): if not isinstance(processing_hooks, (list, tuple)):
for hook in processing_hook: processing_hooks = [processing_hooks]
dic = settings.import_from_str(hook)(dic, user=user)
else: for hook_string in processing_hooks:
dic = settings.import_from_str(processing_hook)(dic, user=user) hook = settings.import_from_str(hook_string)
dic = hook(dic, user=user, scope=scope)
return dic return dic

View file

@ -107,7 +107,7 @@ def fake_sub_generator(user):
return user.email return user.email
def fake_idtoken_processing_hook(id_token, user): def fake_idtoken_processing_hook(id_token, user, scope=None):
""" """
Fake function for inserting some keys into token. Testing OIDC_IDTOKEN_PROCESSING_HOOK. Fake function for inserting some keys into token. Testing OIDC_IDTOKEN_PROCESSING_HOOK.
""" """
@ -116,10 +116,18 @@ def fake_idtoken_processing_hook(id_token, user):
return id_token return id_token
def fake_idtoken_processing_hook2(id_token, user): def fake_idtoken_processing_hook2(id_token, user, scope=None):
""" """
Fake function for inserting some keys into token. Testing OIDC_IDTOKEN_PROCESSING_HOOK - tuple or list as param Fake function for inserting some keys into token. Testing OIDC_IDTOKEN_PROCESSING_HOOK - tuple or list as param
""" """
id_token['test_idtoken_processing_hook2'] = FAKE_RANDOM_STRING id_token['test_idtoken_processing_hook2'] = FAKE_RANDOM_STRING
id_token['test_idtoken_processing_hook_user_email2'] = user.email id_token['test_idtoken_processing_hook_user_email2'] = user.email
return id_token return id_token
def fake_idtoken_processing_hook3(id_token, user, scope=None):
"""
Fake function for checking scope is passed to processing hook.
"""
id_token['scope_passed_to_processing_hook'] = scope
return id_token

View file

@ -734,6 +734,26 @@ class TokenTestCase(TestCase):
self.assertEqual(id_token.get('test_idtoken_processing_hook2'), FAKE_RANDOM_STRING) self.assertEqual(id_token.get('test_idtoken_processing_hook2'), FAKE_RANDOM_STRING)
self.assertEqual(id_token.get('test_idtoken_processing_hook_user_email2'), self.user.email) self.assertEqual(id_token.get('test_idtoken_processing_hook_user_email2'), self.user.email)
@override_settings(
OIDC_IDTOKEN_PROCESSING_HOOK=(
'oidc_provider.tests.app.utils.fake_idtoken_processing_hook3'))
def test_additional_idtoken_processing_hook_scope_param(self):
"""
Test scope parameter is passed to OIDC_IDTOKEN_PROCESSING_HOOK.
"""
code = self._create_code(['openid', 'email', 'profile', 'dummy'])
post_data = self._auth_code_post_data(code=code.code)
response = self._post_request(post_data)
response_dic = json.loads(response.content.decode('utf-8'))
id_token = JWT().unpack(response_dic['id_token'].encode('utf-8')).payload()
self.assertEqual(
id_token.get('scope_passed_to_processing_hook'),
['openid', 'email', 'profile', 'dummy'])
def test_pkce_parameters(self): def test_pkce_parameters(self):
""" """
Test Proof Key for Code Exchange by OAuth Public Clients. Test Proof Key for Code Exchange by OAuth Public Clients.