openstack
/
keystone
Code Issues Proposed changes

Update AuthContextMiddleware to not use token_api 

AuthContextMiddleware and the fall-through in the controller base
_build_policy_check_credentials now utilizes the KeystoneToken
model and uses token_provider_api instead of token_api.

In support of this change, the token auth plugin, the auth_context
builder, token bind check, and token controller have all been updated
to utilize the KeystoneToken model. Support for the federation data
has been added to the KeystoneToken model so that it can be used
in the auth context and associated code.

Associated tests that passed a raw token_ref to methods that now
expect the KeystoneToken model have been updated. This includes an
update to the revocation model to guard against users without domain
data (the federated user case).

Change-Id: I81da15137a0ab3778d835c8de1ec8ed9e5b301f6
bp: non-persistent-tokens
This commit is contained in:
Morgan Fainberg 2014-08-19 17:21:40 -07:00
parent ea185a25a2
commit 37f59f4b0b
12 changed files with 269 additions and 222 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
keystone/auth/plugins/token.py
View File

@ -12,12 +12,11 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo.utils import timeutils
from keystone import auth
from keystone.common import dependency
from keystone.common import wsgi
from keystone import exception
from keystone.models import token_model
from keystone.openstack.common import log
@ -36,21 +35,14 @@ class Token(auth.AuthMethodHandler):
target=self.method)
token_id = auth_payload['id']
response = self.token_provider_api.validate_token(token_id)
# For V3 tokens, the essential data is under the 'token' value.
# For V2, the comparable data was nested under 'access'.
token_ref = response.get('token', response.get('access'))
token_ref = token_model.KeystoneToken(token_id=token_id,
token_data=response)
# Do not allow tokens used for delegation to
# create another token, or perform any changes of
# state in Keystone. To do so is to invite elevation of
# privilege attacks
if 'OS-TRUST:trust' in token_ref:
raise exception.Forbidden()
if 'trust' in token_ref:
raise exception.Forbidden()
if 'trust_id' in token_ref.get('metadata', {}):
raise exception.Forbidden()
if 'OS-OAUTH1' in token_ref:
if token_ref.oauth_scoped or token_ref.trust_scoped:
raise exception.Forbidden()
wsgi.validate_token_bind(context, token_ref)
@ -68,22 +60,13 @@ class Token(auth.AuthMethodHandler):
# issued prior to audit id existing, the chain is not tracked.
token_audit_id = None
# New tokens are not allowed to extend the expiration
# time of an old token, otherwise, they could be extened
# forever. The expiration value was stored at different
# locations in v2 and v3 tokens.
expires_at = token_ref.get('expires_at')
if not expires_at:
expires_at = token_ref.get('expires')
if not expires_at:
expires_at = timeutils.normalize_time(
timeutils.parse_isotime(token_ref['token']['expires']))
user_context.setdefault('expires_at', expires_at)
user_context.setdefault('expires_at', token_ref.expires)
user_context['audit_id'] = token_audit_id
user_context.setdefault('user_id', token_ref['user']['id'])
user_context.setdefault('user_id', token_ref.user_id)
# TODO(morganfainberg: determine if token 'extras' can be removed
# from the user_context
user_context['extras'].update(token_ref.get('extras', {}))
user_context['method_names'].extend(token_ref.get('methods', []))
user_context['method_names'].extend(token_ref.methods)
except AssertionError as e:
LOG.error(e)

127
keystone/common/authorization.py
View File

@ -16,9 +16,9 @@
# License for the specific language governing permissions and limitations
# under the License.
from keystone.contrib import federation
from keystone import exception
from keystone.i18n import _
from keystone.models import token_model
from keystone.openstack.common import log
@ -41,89 +41,46 @@ It is a dictionary with the following attributes:
LOG = log.getLogger(__name__)
def is_v3_token(token):
# V3 token data are encapsulated into "token" key while
# V2 token data are encapsulated into "access" key.
return 'token' in token
def v3_token_to_auth_context(token):
creds = {'is_delegated_auth': False}
token_data = token['token']
try:
creds['user_id'] = token_data['user']['id']
except AttributeError:
LOG.warning(_('RBAC: Invalid user data in v3 token'))
raise exception.Unauthorized()
if 'project' in token_data:
creds['project_id'] = token_data['project']['id']
else:
LOG.debug('RBAC: Proceeding without project')
if 'domain' in token_data:
creds['domain_id'] = token_data['domain']['id']
if 'roles' in token_data:
creds['roles'] = []
for role in token_data['roles']:
creds['roles'].append(role['name'])
creds['group_ids'] = [
g['id'] for g in token_data['user'].get(federation.FEDERATION, {}).get(
'groups', [])]
trust = token_data.get('OS-TRUST:trust')
if trust is None:
creds['trust_id'] = None
creds['trustor_id'] = None
creds['trustee_id'] = None
else:
creds['trust_id'] = trust['id']
creds['trustor_id'] = trust['trustor_user']['id']
creds['trustee_id'] = trust['trustee_user']['id']
creds['is_delegated_auth'] = True
oauth1 = token_data.get('OS-OAUTH1')
if oauth1 is None:
creds['consumer_id'] = None
creds['access_token_id'] = None
else:
creds['consumer_id'] = oauth1['consumer_id']
creds['access_token_id'] = oauth1['access_token_id']
creds['is_delegated_auth'] = True
return creds
def v2_token_to_auth_context(token):
creds = {'is_delegated_auth': False}
token_data = token['access']
try:
creds['user_id'] = token_data['user']['id']
except AttributeError:
LOG.warning(_('RBAC: Invalid user data in v2 token'))
raise exception.Unauthorized()
if 'tenant' in token_data['token']:
creds['project_id'] = token_data['token']['tenant']['id']
else:
LOG.debug('RBAC: Proceeding without tenant')
if 'roles' in token_data['user']:
creds['roles'] = [role['name'] for
role in token_data['user']['roles']]
trust = token_data.get('trust')
if trust is None:
creds['trust_id'] = None
creds['trustor_id'] = None
creds['trustee_id'] = None
else:
creds['trust_id'] = trust.get('id')
creds['trustor_id'] = trust.get('trustor_id')
creds['trustee_id'] = trust.get('trustee_id')
creds['is_delegated_auth'] = True
return creds
def token_to_auth_context(token):
if is_v3_token(token):
creds = v3_token_to_auth_context(token)
if not isinstance(token, token_model.KeystoneToken):
raise exception.UnexpectedError(_('token reference must be a '
'KeystoneToken type, got: %s') %
type(token))
auth_context = {'token': token,
'is_delegated_auth': False}
try:
auth_context['user_id'] = token.user_id
except KeyError:
LOG.warning(_('RBAC: Invalid user data in token'))
raise exception.Unauthorized()
if token.project_scoped:
auth_context['project_id'] = token.project_id
elif token.domain_scoped:
auth_context['domain_id'] = token.domain_id
else:
creds = v2_token_to_auth_context(token)
return creds
LOG.debug('RBAC: Proceeding without project or domain scope')
if token.trust_scoped:
auth_context['is_delegated_auth'] = True
auth_context['trust_id'] = token.trust_id
auth_context['trustor_id'] = token.trustor_user_id
auth_context['trustee_id'] = token.trustee_user_id
else:
auth_context['trust_id'] = None
auth_context['trustor_id'] = None
auth_context['trustee_id'] = None
roles = token.role_names
if roles:
auth_context['roles'] = roles
if token.oauth_scoped:
auth_context['is_delegated_auth'] = True
auth_context['consumer_id'] = token.oauth_consumer_id
auth_context['access_token_id'] = token.oauth_access_token_id
if token.is_federated_user:
auth_context['group_ids'] = token.federation_group_ids
return auth_context

27
keystone/common/controller.py
View File

@ -23,6 +23,7 @@ from keystone.common import wsgi
from keystone import config
from keystone import exception
from keystone.i18n import _
from keystone.models import token_model
from keystone.openstack.common import log
@ -59,26 +60,24 @@ def _build_policy_check_credentials(self, action, context, kwargs):
LOG.debug('RBAC: using auth context from the request environment')
return context['environment'].get(authorization.AUTH_CONTEXT_ENV)
# now build the auth context from the incoming auth token
# There is no current auth context, build it from the incoming token.
# TODO(morganfainberg): Collapse this logic with AuthContextMiddleware
# in sane manner as this just mirrors the logic in AuthContextMiddleware
try:
LOG.debug('RBAC: building auth context from the incoming auth token')
# TODO(ayoung): These two functions return the token in different
# formats. However, the call
# to get_token hits the caching layer, and does not validate the
# token. This should be reduced to one call
if not CONF.token.revoke_by_id:
self.token_api.token_provider_api.validate_token(
context['token_id'])
token_ref = self.token_api.get_token(context['token_id'])
token_ref = token_model.KeystoneToken(
token_id=context['token_id'],
token_data=self.token_provider_api.validate_token(
context['token_id']))
# NOTE(jamielennox): whilst this maybe shouldn't be within this
# function it would otherwise need to reload the token_ref from
# backing store.
wsgi.validate_token_bind(context, token_ref)
except exception.TokenNotFound:
LOG.warning(_('RBAC: Invalid token'))
raise exception.Unauthorized()
# NOTE(jamielennox): whilst this maybe shouldn't be within this function
# it would otherwise need to reload the token_ref from backing store.
wsgi.validate_token_bind(context, token_ref)
auth_context = authorization.token_to_auth_context(token_ref['token_data'])
auth_context = authorization.token_to_auth_context(token_ref)
return auth_context

29
keystone/common/wsgi.py
View File

@ -18,6 +18,8 @@
"""Utility methods for working with WSGI servers."""
import copy
from oslo import i18n
import routes.middleware
import six
@ -52,7 +54,11 @@ def validate_token_bind(context, token_ref):
if bind_mode == 'disabled':
return
bind = token_ref.get('bind', {})
if not isinstance(token_ref, token_model.KeystoneToken):
raise exception.UnexpectedError(_('token reference must be a '
'KeystoneToken type, got: %s') %
type(token_ref))
bind = token_ref.bind
# permissive and strict modes don't require there to be a bind
permissive = bind_mode in ('permissive', 'strict')
@ -264,28 +270,29 @@ class Application(BaseApplication):
def assert_admin(self, context):
if not context['is_admin']:
try:
user_token_ref = self.token_api.get_token(context['token_id'])
user_token_ref = token_model.KeystoneToken(
token_id=context['token_id'],
token_data=self.token_provider_api.validate_token(
context['token_id']))
except exception.TokenNotFound as e:
raise exception.Unauthorized(e)
validate_token_bind(context, user_token_ref)
creds = user_token_ref['metadata'].copy()
creds = copy.deepcopy(user_token_ref.metadata)
try:
creds['user_id'] = user_token_ref['user'].get('id')
except AttributeError:
creds['user_id'] = user_token_ref.user_id
except exception.UnexpectedError:
LOG.debug('Invalid user')
raise exception.Unauthorized()
try:
creds['tenant_id'] = user_token_ref['tenant'].get('id')
except AttributeError:
if user_token_ref.project_scoped:
creds['tenant_id'] = user_token_ref.project_id
else:
LOG.debug('Invalid tenant')
raise exception.Unauthorized()
# NOTE(vish): this is pretty inefficient
creds['roles'] = [self.assignment_api.get_role(role)['name']
for role in creds.get('roles', [])]
creds['roles'] = user_token_ref.role_names
# Accept either is_admin or the admin role
self.policy_api.enforce(creds, 'admin_required', {})

4
keystone/contrib/revoke/model.py
View File

@ -318,7 +318,9 @@ def build_token_values(token_data):
user = token_data.get('user')
if user is not None:
token_values['user_id'] = user['id']
token_values['identity_domain_id'] = user['domain']['id']
# Federated users do not have a domain, be defensive and get the user
# domain set to None in the federated user case.
token_values['identity_domain_id'] = user.get('domain', {}).get('id')
else:
token_values['user_id'] = None
token_values['identity_domain_id'] = None

16
keystone/middleware/core.py
View File

@ -22,6 +22,7 @@ from keystone.common import utils
from keystone.common import wsgi
from keystone import exception
from keystone.i18n import _
from keystone.models import token_model
from keystone.openstack.common import jsonutils
from keystone.openstack.common import log
from keystone.openstack.common import versionutils
@ -253,20 +254,13 @@ class AuthContextMiddleware(wsgi.Middleware):
context['environment'] = request.environ
try:
token_ref = self.token_api.get_token(token_id)
# TODO(ayoung): These two functions return the token in different
# formats instead of two calls, only make one. However, the call
# to get_token hits the caching layer, and does not validate the
# token. In the future, this should be reduced to one call.
if not CONF.token.revoke_by_id:
self.token_api.token_provider_api.validate_token(
context['token_id'])
token_ref = token_model.KeystoneToken(
token_id=token_id,
token_data=self.token_provider_api.validate_token(token_id))
# TODO(gyee): validate_token_bind should really be its own
# middleware
wsgi.validate_token_bind(context, token_ref)
return authorization.token_to_auth_context(
token_ref['token_data'])
return authorization.token_to_auth_context(token_ref)
except exception.TokenNotFound:
LOG.warning(_('RBAC: Invalid token'))
raise exception.Unauthorized()

57
keystone/models/token_model.py
View File

@ -12,10 +12,12 @@
"""Unified in-memory token model."""
from keystoneclient.common import cms
from oslo.utils import timeutils
import six
from keystone.common import config
from keystone.contrib import federation
from keystone import exception
from keystone.i18n import _
@ -52,6 +54,8 @@ class KeystoneToken(dict):
else:
raise exception.UnsupportedTokenVersionException()
self.token_id = token_id
self.short_id = cms.cms_hash_token(token_id,
mode=CONF.token.hash_algorithm)
if self.project_scoped and self.domain_scoped:
raise exception.UnexpectedError(_('Found invalid token: scoped to '
@ -238,16 +242,20 @@ class KeystoneToken(dict):
else:
return self.get('trust', {}).get('trustor_user_id')
@property
def oauth_scoped(self):
return 'OS-OAUTH1' in self
@property
def oauth_access_token_id(self):
if self.version is V3:
return self.get('OS-OAUTH1', {}).get('access_token_id')
if self.version is V3 and self.oauth_scoped:
return self['OS-OAUTH1']['access_token_id']
return None
@property
def oauth_consumer_id(self):
if self.version is V3:
return self.get('OS-OAUTH1', {}).get('consumer_id')
if self.version is V3 and self.oauth_scoped:
return self['OS-OAUTH1']['consumer_id']
return None
@property
@ -269,3 +277,44 @@ class KeystoneToken(dict):
if self.version is V3:
return self.get('bind')
return self.get('token', {}).get('bind')
@property
def is_federated_user(self):
try:
return self.version is V3 and federation.FEDERATION in self['user']
except KeyError:
raise exception.UnexpectedError()
@property
def federation_group_ids(self):
if self.is_federated_user:
if self.version is V3:
try:
groups = self['user'][federation.FEDERATION].get(
'groups', [])
return [g['id'] for g in groups]
except KeyError:
raise exception.UnexpectedError()
return []
@property
def federation_idp_id(self):
if self.version is not V3 or not self.is_federated_user:
return None
return self['user'][federation.FEDERATION]['identity_provider']['id']
@property
def federation_protocol_id(self):
if self.version is V3 and self.is_federated_user:
return self['user'][federation.FEDERATION]['protocol']['id']
return None
@property
def metadata(self):
return self.get('metadata', {})
@property
def methods(self):
if self.version is V3:
return self.get('methods', [])
return []

8
keystone/tests/test_auth.py
View File

@ -26,6 +26,7 @@ from keystone.common import authorization
from keystone.common import environment
from keystone import config
from keystone import exception
from keystone.models import token_model
from keystone import tests
from keystone.tests import default_fixtures
from keystone.tests.ksfixtures import database
@ -811,9 +812,10 @@ class AuthWithTrust(AuthTest):
self.config_fixture.config(group='trust', enabled=True)
def _create_auth_context(self, token_id):
token_ref = self.token_api.get_token(token_id)
auth_context = authorization.token_to_auth_context(
token_ref['token_data'])
token_ref = token_model.KeystoneToken(
token_id=token_id,
token_data=self.token_provider_api.validate_token(token_id))
auth_context = authorization.token_to_auth_context(token_ref)
return {'environment': {authorization.AUTH_CONTEXT_ENV: auth_context},
'token_id': token_id,
'host_url': HOST_URL}

13
keystone/tests/test_content_types.py
View File

@ -13,6 +13,7 @@
# under the License.
import json
import time
import uuid
from keystoneclient.common import cms
@ -1216,7 +1217,19 @@ class JsonTestCase(RestfulTestCase, CoreApiTests, LegacyV2UsernameTests):
token1 = self.get_scoped_token()
# TODO(morganfainberg): Because this is making a restful call to the
# app a change to UTCNOW via mock.patch will not affect the returned
# token. The only surefire way to ensure there is not a transient bug
# based upon when the second token is issued is with a sleep. This
# issue all stems from the limited resolution (no microseconds) on the
# expiry time of tokens and the way revocation events utilizes token
# expiry to revoke individual tokens. This is a stop-gap until all
# associated issues with resolution on expiration and revocation events
# are resolved.
time.sleep(1)
token2 = self.get_scoped_token()
self.admin_request(method='DELETE',
path='/v2.0/tokens/%s' % token2,
token=token1)

71
keystone/tests/test_token_bind.py
View File

@ -12,20 +12,18 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
import uuid
from keystone.common import wsgi
from keystone import exception
from keystone.models import token_model
from keystone import tests
from keystone.tests import test_token_provider
KERBEROS_BIND = 'USER@REALM'
# the only thing the function checks for is the presence of bind
TOKEN_BIND_KERB = {'bind': {'kerberos': KERBEROS_BIND}}
TOKEN_BIND_UNKNOWN = {'bind': {'FOO': 'BAR'}}
TOKEN_BIND_NONE = {}
ANY = 'any'
ALL_TOKENS = [TOKEN_BIND_KERB, TOKEN_BIND_UNKNOWN, TOKEN_BIND_NONE]
class BindTest(tests.TestCase):
@ -35,6 +33,20 @@ class BindTest(tests.TestCase):
will apply to all future binding mechanisms.
"""
def setUp(self):
super(BindTest, self).setUp()
self.TOKEN_BIND_KERB = copy.deepcopy(
test_token_provider.SAMPLE_V3_TOKEN)
self.TOKEN_BIND_KERB['token']['bind'] = {'kerberos': KERBEROS_BIND}
self.TOKEN_BIND_UNKNOWN = copy.deepcopy(
test_token_provider.SAMPLE_V3_TOKEN)
self.TOKEN_BIND_UNKNOWN['token']['bind'] = {'FOO': 'BAR'}
self.TOKEN_BIND_NONE = copy.deepcopy(
test_token_provider.SAMPLE_V3_TOKEN)
self.ALL_TOKENS = [self.TOKEN_BIND_KERB, self.TOKEN_BIND_UNKNOWN,
self.TOKEN_BIND_NONE]
def assert_kerberos_bind(self, tokens, bind_level,
use_kerberos=True, success=True):
if not isinstance(tokens, dict):
@ -55,17 +67,22 @@ class BindTest(tests.TestCase):
context['environment']['REMOTE_USER'] = KERBEROS_BIND
context['environment']['AUTH_TYPE'] = 'Negotiate'
# NOTE(morganfainberg): This assumes a V3 token.
token_ref = token_model.KeystoneToken(
token_id=uuid.uuid4().hex,
token_data=tokens)
if not success:
self.assertRaises(exception.Unauthorized,
wsgi.validate_token_bind,
context, tokens)
context, token_ref)
else:
wsgi.validate_token_bind(context, tokens)
wsgi.validate_token_bind(context, token_ref)
# DISABLED
def test_bind_disabled_with_kerb_user(self):
self.assert_kerberos_bind(ALL_TOKENS,
self.assert_kerberos_bind(self.ALL_TOKENS,
bind_level='disabled',
use_kerberos=ANY,
success=True)
@ -73,25 +90,25 @@ class BindTest(tests.TestCase):
# PERMISSIVE
def test_bind_permissive_with_kerb_user(self):
self.assert_kerberos_bind(TOKEN_BIND_KERB,
self.assert_kerberos_bind(self.TOKEN_BIND_KERB,
bind_level='permissive',
use_kerberos=True,
success=True)
def test_bind_permissive_with_regular_token(self):
self.assert_kerberos_bind(TOKEN_BIND_NONE,
self.assert_kerberos_bind(self.TOKEN_BIND_NONE,
bind_level='permissive',
use_kerberos=ANY,
success=True)
def test_bind_permissive_without_kerb_user(self):
self.assert_kerberos_bind(TOKEN_BIND_KERB,
self.assert_kerberos_bind(self.TOKEN_BIND_KERB,
bind_level='permissive',
use_kerberos=False,
success=False)
def test_bind_permissive_with_unknown_bind(self):
self.assert_kerberos_bind(TOKEN_BIND_UNKNOWN,
self.assert_kerberos_bind(self.TOKEN_BIND_UNKNOWN,
bind_level='permissive',
use_kerberos=ANY,
success=True)
@ -99,25 +116,25 @@ class BindTest(tests.TestCase):
# STRICT
def test_bind_strict_with_regular_token(self):
self.assert_kerberos_bind(TOKEN_BIND_NONE,
self.assert_kerberos_bind(self.TOKEN_BIND_NONE,
bind_level='strict',
use_kerberos=ANY,
success=True)
def test_bind_strict_with_kerb_user(self):
self.assert_kerberos_bind(TOKEN_BIND_KERB,
self.assert_kerberos_bind(self.TOKEN_BIND_KERB,
bind_level='strict',
use_kerberos=True,
success=True)
def test_bind_strict_without_kerb_user(self):
self.assert_kerberos_bind(TOKEN_BIND_KERB,
self.assert_kerberos_bind(self.TOKEN_BIND_KERB,
bind_level='strict',
use_kerberos=False,
success=False)
def test_bind_strict_with_unknown_bind(self):
self.assert_kerberos_bind(TOKEN_BIND_UNKNOWN,
self.assert_kerberos_bind(self.TOKEN_BIND_UNKNOWN,
bind_level='strict',
use_kerberos=ANY,
success=False)
@ -125,25 +142,25 @@ class BindTest(tests.TestCase):
# REQUIRED
def test_bind_required_with_regular_token(self):
self.assert_kerberos_bind(TOKEN_BIND_NONE,
self.assert_kerberos_bind(self.TOKEN_BIND_NONE,
bind_level='required',
use_kerberos=ANY,
success=False)
def test_bind_required_with_kerb_user(self):
self.assert_kerberos_bind(TOKEN_BIND_KERB,
self.assert_kerberos_bind(self.TOKEN_BIND_KERB,
bind_level='required',
use_kerberos=True,
success=True)
def test_bind_required_without_kerb_user(self):
self.assert_kerberos_bind(TOKEN_BIND_KERB,
self.assert_kerberos_bind(self.TOKEN_BIND_KERB,
bind_level='required',
use_kerberos=False,
success=False)
def test_bind_required_with_unknown_bind(self):
self.assert_kerberos_bind(TOKEN_BIND_UNKNOWN,
self.assert_kerberos_bind(self.TOKEN_BIND_UNKNOWN,
bind_level='required',
use_kerberos=ANY,
success=False)
@ -151,31 +168,31 @@ class BindTest(tests.TestCase):
# NAMED
def test_bind_named_with_regular_token(self):
self.assert_kerberos_bind(TOKEN_BIND_NONE,
self.assert_kerberos_bind(self.TOKEN_BIND_NONE,
bind_level='kerberos',
use_kerberos=ANY,
success=False)
def test_bind_named_with_kerb_user(self):
self.assert_kerberos_bind(TOKEN_BIND_KERB,
self.assert_kerberos_bind(self.TOKEN_BIND_KERB,
bind_level='kerberos',
use_kerberos=True,
success=True)
def test_bind_named_without_kerb_user(self):
self.assert_kerberos_bind(TOKEN_BIND_KERB,
self.assert_kerberos_bind(self.TOKEN_BIND_KERB,
bind_level='kerberos',
use_kerberos=False,
success=False)
def test_bind_named_with_unknown_bind(self):
self.assert_kerberos_bind(TOKEN_BIND_UNKNOWN,
self.assert_kerberos_bind(self.TOKEN_BIND_UNKNOWN,
bind_level='kerberos',
use_kerberos=ANY,
success=False)
def test_bind_named_with_unknown_scheme(self):
self.assert_kerberos_bind(ALL_TOKENS,
self.assert_kerberos_bind(self.ALL_TOKENS,
bind_level='unknown',
use_kerberos=ANY,
success=False)

43
keystone/tests/unit/token/test_token_model.py
View File

@ -114,6 +114,49 @@ class TestKeystoneTokenModel(core.TestCase):
self.assertIsNone(token_data.audit_id)
self.assertIsNone(token_data.audit_chain_id)
def test_token_model_v3_federated_user(self):
token_data = token_model.KeystoneToken(token_id=uuid.uuid4().hex,
token_data=self.v3_sample_token)
federation_data = {'identity_provider': {'id': uuid.uuid4().hex},
'protocol': {'id': 'saml2'},
'groups': [{'id': uuid.uuid4().hex}
for x in range(1, 5)]}
self.assertFalse(token_data.is_federated_user)
self.assertEqual([], token_data.federation_group_ids)
self.assertIsNone(token_data.federation_protocol_id)
self.assertIsNone(token_data.federation_idp_id)
token_data['user'][token_model.federation.FEDERATION] = federation_data
self.assertTrue(token_data.is_federated_user)
self.assertEqual([x['id'] for x in federation_data['groups']],
token_data.federation_group_ids)
self.assertEqual(federation_data['protocol']['id'],
token_data.federation_protocol_id)
self.assertEqual(federation_data['identity_provider']['id'],
token_data.federation_idp_id)
def test_token_model_v2_federated_user(self):
token_data = token_model.KeystoneToken(token_id=uuid.uuid4().hex,
token_data=self.v2_sample_token)
federation_data = {'identity_provider': {'id': uuid.uuid4().hex},
'protocol': {'id': 'saml2'},
'groups': [{'id': uuid.uuid4().hex}
for x in range(1, 5)]}
self.assertFalse(token_data.is_federated_user)
self.assertEqual([], token_data.federation_group_ids)
self.assertIsNone(token_data.federation_protocol_id)
self.assertIsNone(token_data.federation_idp_id)
token_data['user'][token_model.federation.FEDERATION] = federation_data
# Federated users should not exist in V2, the data should remain empty
self.assertFalse(token_data.is_federated_user)
self.assertEqual([], token_data.federation_group_ids)
self.assertIsNone(token_data.federation_protocol_id)
self.assertIsNone(token_data.federation_idp_id)
def test_token_model_v2(self):
token_data = token_model.KeystoneToken(uuid.uuid4().hex,
self.v2_sample_token)

61
keystone/token/controllers.py
View File

@ -25,6 +25,7 @@ from keystone.common import wsgi
from keystone import config
from keystone import exception
from keystone.i18n import _
from keystone.models import token_model
from keystone.openstack.common import jsonutils
from keystone.openstack.common import log
from keystone.token import provider
@ -40,7 +41,7 @@ class ExternalAuthNotApplicable(Exception):
@dependency.requires('assignment_api', 'catalog_api', 'identity_api',
'token_api', 'token_provider_api', 'trust_api')
'token_provider_api', 'trust_api')
class Auth(controller.V2Controller):
@controller.v2_deprecated
@ -169,20 +170,19 @@ class Auth(controller.V2Controller):
size=CONF.max_token_size)
try:
old_token_ref = self.token_api.get_token(old_token)
token_model_ref = token_model.KeystoneToken(
token_id=old_token,
token_data=self.token_provider_api.validate_token(old_token))
except exception.NotFound as e:
raise exception.Unauthorized(e)
wsgi.validate_token_bind(context, old_token_ref)
wsgi.validate_token_bind(context, token_model_ref)
# A trust token cannot be used to get another token
if 'trust' in old_token_ref:
raise exception.Forbidden()
if 'trust_id' in old_token_ref['metadata']:
if token_model_ref.trust_scoped:
raise exception.Forbidden()
user_ref = old_token_ref['user']
user_id = user_ref['id']
user_id = token_model_ref.user_id
tenant_id = self._get_project_id_from_auth(auth)
if not CONF.trust.enabled and 'trust_id' in auth:
@ -222,7 +222,7 @@ class Auth(controller.V2Controller):
tenant_ref, metadata_ref['roles'] = self._get_project_roles_and_ref(
user_id, tenant_id)
expiry = old_token_ref['expires']
expiry = token_model_ref.expires
if CONF.trust.enabled and 'trust_id' in auth:
trust_id = auth['trust_id']
trust_roles = []
@ -241,29 +241,8 @@ class Auth(controller.V2Controller):
metadata_ref['trustee_user_id'] = trust_ref['trustee_user_id']
metadata_ref['trust_id'] = trust_id
bind = old_token_ref.get('bind')
# TODO(morganfainberg): Convert this over to using the KeystoneToken
# model when removing dependency on token_api.
token_data = old_token_ref.get('token_data')
audit_id = None
if token_data:
# NOTE(morganfainberg): The token audit field will always contain
# the token's direct audit id at index 0, index 1 will exist and
# contain the audit chain id (audit id of the original token in
# the chain), so always lookup the last element of the audit field
# to determine the id to pass on.
try:
if 'access' in token_data:
audit_id = token_data['access']['token'].get(
'audit_ids', [])[-1]
else:
audit_id = token_data['token'].get('audit_ids', [])[-1]
except IndexError:
# NOTE(morganfainberg): When transitioning from tokens without
# audit_ids to tokens with audit ids it some tokens may not
# have an audit_id, and the lookup will cause an IndexError
# to be raised.
pass
bind = token_model_ref.bind
audit_id = token_model_ref.audit_chain_id
return (current_user_ref, tenant_ref, metadata_ref, expiry, bind,
audit_id)
@ -422,15 +401,17 @@ class Auth(controller.V2Controller):
Optionally, limited to a token owned by a specific tenant.
"""
data = self.token_api.get_token(token_id)
token_ref = token_model.KeystoneToken(
token_id=token_id,
token_data=self.token_provider_api.validate_token(token_id))
if belongs_to:
if data.get('tenant') is None:
if not token_ref.project_scoped:
raise exception.Unauthorized(
_('Token does not belong to specified tenant.'))
if data['tenant'].get('id') != belongs_to:
if token_ref.project_id != belongs_to:
raise exception.Unauthorized(
_('Token does not belong to specified tenant.'))
return data
return token_ref
@controller.v2_deprecated
@controller.protected()
@ -497,11 +478,11 @@ class Auth(controller.V2Controller):
token_ref = self._get_token_ref(token_id)
catalog_ref = None
if token_ref.get('tenant'):
if token_ref.project_id:
catalog_ref = self.catalog_api.get_catalog(
token_ref['user']['id'],
token_ref['tenant']['id'],
token_ref['metadata'])
token_ref.user_id,
token_ref.project_id,
token_ref.metadata)
return Auth.format_endpoint_list(catalog_ref)