kumi/django-oidc-provider
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Fix scope handling of token endpoint

Browse source 
The token endpoint handled the scope parameter incorrectly for all of
the three handled grant types:

 1. For "authorization_code" grant type the scope parameter in the token
    request should not be respected but the scope should be taken from
    the authorization code.  It was not totally ignored, but rather the
    scope parameter of the token request was used for the generated ID
    token.  This had two consequences:

      * Spec conforming implementations of authorization code flow
        didn't get correct ID tokens, since they usually don't pass
        scope parameter with the token request.

      * It's possible to get a broader scope for the ID token than what
        is authorized by the user in the original authorization code
        request.

 2. For "refresh_token" grant type the scope parameter in the token
    request should only allow narrowing down the scope.  It wasn't
    narrowed, but rather the original auth code scope was used for the
    access token and the passed in scope parameter was used for the ID
    token (again allowing unauthorized scopes in the ID token).

 3. For "password" grant type the scope parameter in the token request
    should be respected.  The problem with this was that it wasn't
    properly splitted when passed to ID token creation.

Fixes #186
This commit is contained in:
Tuomas Suutari 2017-07-07 21:58:05 +03:00
parent 1397439b09
commit 6199a9a17e
3 changed files with 116 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
oidc_provider/lib/endpoints/token.py
View file

@ -162,6 +162,8 @@ class TokenEndpoint(object):
return self.create_access_token_response_dic()
def create_access_token_response_dic(self):
# See https://tools.ietf.org/html/rfc6749#section-4.3
token = create_token(
self.user,
self.client,
@ -173,7 +175,7 @@ class TokenEndpoint(object):
nonce='self.code.nonce',
at_hash=token.at_hash,
request=self.request,
scope=self.params['scope'],
scope=token.scope,
)
token.id_token = id_token_dic
@ -188,6 +190,8 @@ class TokenEndpoint(object):
}
def create_code_response_dic(self):
# See https://tools.ietf.org/html/rfc6749#section-4.1
token = create_token(
user=self.code.user,
client=self.code.client,
@ -200,7 +204,7 @@ class TokenEndpoint(object):
nonce=self.code.nonce,
at_hash=token.at_hash,
request=self.request,
scope=self.params['scope'],
scope=token.scope,
)
else:
id_token_dic = {}
@ -223,10 +227,18 @@ class TokenEndpoint(object):
return dic
def create_refresh_response_dic(self):
# See https://tools.ietf.org/html/rfc6749#section-6
scope_param = self.params['scope']
scope = (scope_param.split(' ') if scope_param else self.token.scope)
unauthorized_scopes = set(scope) - set(self.token.scope)
if unauthorized_scopes:
raise TokenError('invalid_scope')
token = create_token(
user=self.token.user,
client=self.token.client,
scope=self.token.scope)
scope=scope)
# If the Token has an id_token it's an Authentication request.
if self.token.id_token:
@ -236,7 +248,7 @@ class TokenEndpoint(object):
nonce=None,
at_hash=token.at_hash,
request=self.request,
scope=self.params['scope'],
scope=token.scope,
)
else:
id_token_dic = {}

2
oidc_provider/tests/app/utils.py
View file

@ -29,6 +29,8 @@ def create_fake_user():
user = User()
user.username = 'johndoe'
user.email = 'johndoe@example.com'
user.first_name = 'John'
user.last_name = 'Doe'
user.set_password('1234')
user.save()

115
oidc_provider/tests/test_token_endpoint.py
View file

@ -50,15 +50,18 @@ class TokenTestCase(TestCase):
self.user = create_fake_user()
self.client = create_fake_client(response_type='code')
def _password_grant_post_data(self):
return {
def _password_grant_post_data(self, scope=None):
result = {
'username': 'johndoe',
'password': '1234',
'grant_type': 'password',
'scope': 'openid email',
}
if scope is not None:
result['scope'] = ' '.join(scope)
return result
def _auth_code_post_data(self, code):
def _auth_code_post_data(self, code, scope=None):
"""
All the data that will be POSTed to the Token Endpoint.
"""
@ -70,10 +73,12 @@ class TokenTestCase(TestCase):
'code': code,
'state': uuid.uuid4().hex,
}
if scope is not None:
post_data['scope'] = ' '.join(scope)
return post_data
def _refresh_token_post_data(self, refresh_token):
def _refresh_token_post_data(self, refresh_token, scope=None):
"""
All the data that will be POSTed to the Token Endpoint.
"""
@ -83,6 +88,8 @@ class TokenTestCase(TestCase):
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
}
if scope is not None:
post_data['scope'] = ' '.join(scope)
return post_data
@ -103,14 +110,14 @@ class TokenTestCase(TestCase):
return response
def _create_code(self):
def _create_code(self, scope=None):
"""
Generate a valid grant code.
"""
code = create_code(
user=self.user,
client=self.client,
scope=['openid', 'email'],
scope=(scope if scope else ['openid', 'email']),
nonce=FAKE_NONCE,
is_authentication=True)
code.save()
@ -228,30 +235,41 @@ class TokenTestCase(TestCase):
self.assertEqual(400, response.status_code)
self.assertEqual('invalid_client', response_dict['error'])
@patch('oidc_provider.lib.utils.token.uuid')
def test_password_grant_full_response(self):
self.check_password_grant(scope=['openid', 'email'])
def test_password_grant_scope(self):
self.check_password_grant(scope=['openid', 'profile'])
@override_settings(OIDC_TOKEN_EXPIRE=120,
OIDC_GRANT_TYPE_PASSWORD_ENABLE=True)
def test_password_grant_full_response(self, mock_uuid):
test_hex = 'fake_token'
mock_uuid4 = Mock(spec=uuid.uuid4)
mock_uuid4.hex = test_hex
mock_uuid.uuid4.return_value = mock_uuid4
def check_password_grant(self, scope):
response = self._post_request(
post_data=self._password_grant_post_data(),
post_data=self._password_grant_post_data(scope),
extras=self._password_grant_auth_header()
)
response_dict = json.loads(response.content.decode('utf-8'))
id_token = JWS().verify_compact(response_dict['id_token'].encode('utf-8'), self._get_keys())
self.assertEqual(response_dict['access_token'], 'fake_token')
self.assertEqual(response_dict['refresh_token'], 'fake_token')
token = Token.objects.get(user=self.user)
self.assertEqual(response_dict['access_token'], token.access_token)
self.assertEqual(response_dict['refresh_token'], token.refresh_token)
self.assertEqual(response_dict['expires_in'], 120)
self.assertEqual(response_dict['token_type'], 'bearer')
self.assertEqual(id_token['sub'], str(self.user.id))
self.assertEqual(id_token['aud'], self.client.client_id)
# Check the scope is honored by checking the claims in the userinfo
userinfo_response = self._get_userinfo(response_dict['access_token'])
userinfo = json.loads(userinfo_response.content.decode('utf-8'))
for (scope_param, claim) in [('email', 'email'), ('profile', 'name')]:
if scope_param in scope:
self.assertIn(claim, userinfo)
else:
self.assertNotIn(claim, userinfo)
@override_settings(OIDC_TOKEN_EXPIRE=720)
def test_authorization_code(self):
"""
@ -277,16 +295,64 @@ class TokenTestCase(TestCase):
self.assertEqual(id_token['sub'], str(self.user.id))
self.assertEqual(id_token['aud'], self.client.client_id)
@override_settings(OIDC_TOKEN_EXPIRE=720)
def test_scope_is_ignored_for_auth_code(self):
"""
Scope is ignored for token respones to auth code grant type.
"""
SIGKEYS = self._get_keys()
for code_scope in [['openid'], ['openid', 'email']]:
code = self._create_code(code_scope)
post_data = self._auth_code_post_data(
code=code.code, scope=['openid', 'profile'])
response = self._post_request(post_data)
response_dic = json.loads(response.content.decode('utf-8'))
self.assertEqual(response.status_code, 200)
id_token = JWS().verify_compact(response_dic['id_token'].encode('utf-8'), SIGKEYS)
if 'email' in code_scope:
self.assertIn('email', id_token)
else:
self.assertNotIn('email', id_token)
def test_refresh_token(self):
"""
A request to the Token Endpoint can also use a Refresh Token
by using the grant_type value refresh_token, as described in
Section 6 of OAuth 2.0 [RFC6749].
"""
self.do_refresh_token_check()
def test_refresh_token_invalid_scope(self):
"""
Extending scope in refresh token is not allowed.
Try to get a refresh token with "profile" in the scope even
though the original authorized scope in the authorization code
request is only ['openid', 'email'].
"""
self.do_refresh_token_check(scope=['openid', 'profile'])
def test_refresh_token_narrowed_scope(self):
"""
Narrowing scope in refresh token is allowed.
Try to get a refresh token with just "openid" in the scope even
though the original authorized scope in the authorization code
request is ['openid', 'email'].
"""
self.do_refresh_token_check(scope=['openid'])
def do_refresh_token_check(self, scope=None):
SIGKEYS = self._get_keys()
# Retrieve refresh token
code = self._create_code()
self.assertEqual(code.scope, ['openid', 'email'])
post_data = self._auth_code_post_data(code=code.code)
start_time = time.time()
with patch('oidc_provider.lib.utils.token.time.time') as time_func:
@ -297,14 +363,29 @@ class TokenTestCase(TestCase):
id_token1 = JWS().verify_compact(response_dic1['id_token'].encode('utf-8'), SIGKEYS)
# Use refresh token to obtain new token
post_data = self._refresh_token_post_data(response_dic1['refresh_token'])
post_data = self._refresh_token_post_data(
response_dic1['refresh_token'], scope)
with patch('oidc_provider.lib.utils.token.time.time') as time_func:
time_func.return_value = start_time + 600
response = self._post_request(post_data)
response_dic2 = json.loads(response.content.decode('utf-8'))
if scope and set(scope) - set(code.scope): # too broad scope
self.assertEqual(response.status_code, 400) # Bad Request
self.assertIn('error', response_dic2)
self.assertEqual(response_dic2['error'], 'invalid_scope')
return # No more checks
id_token2 = JWS().verify_compact(response_dic2['id_token'].encode('utf-8'), SIGKEYS)
if scope and 'email' not in scope: # narrowed scope The auth
# The auth code request had email in scope, so it should be
# in the first id token
self.assertIn('email', id_token1)
# but the refresh request had no email in scope
self.assertNotIn('email', id_token2, 'email was not requested')
self.assertNotEqual(response_dic1['id_token'], response_dic2['id_token'])
self.assertNotEqual(response_dic1['access_token'], response_dic2['access_token'])
self.assertNotEqual(response_dic1['refresh_token'], response_dic2['refresh_token'])