openstack
/
keystoneauth
Code Issues Proposed changes

Fix stable/liberty branch 

stable/liberty was failing with multiple issues, so these 3 are
squashed.

54c0c6a fixes this error seen starting glance-api:

 AttributeError: 'Opt' object has no attribute 'group'

b2484fd is a prereq for 0591868

0591868 fixes this error seen starting glance-api:

 AttributeError: 'AccessInfoV3' object has no attribute 'bind'

Fix deprecated options in oslo_config

Converting a keystoneauth Opt to an oslo_config Opt fails because we
convert the deprecated options to a real oslo_config Opt rather than a
DeprecatedOpt.

Orig-Change-Id: I1c86bec7cddcc5751e6584b381e60115b84e1d27
Closes-Bug: #1505906
(cherry picked from commit 54c0c6abe6)

Copy AccessInfo tests from keystoneclient

There were some basic small issues with AccessInfo accessors and it
appears that the tests were never transferred across from
keystoneclient.

Copy those tests as closely as possible.

Orig-Change-Id: I391bf23097c5a8a176a50a938c04fa259df1de12
(cherry picked from commit b2484fdbf6)

Expose bind data via AccessInfo

The bind information is a standard part of the token data and can be
access from auth_token middleware so it should be exposed as part of the
AccessInfo object.

Change-Id: I45fc6eeed43f335aa1d771bdf1a11257432cb85c
(cherry picked from commit 4fd8531fd5)
This commit is contained in:
Jamie Lennox 2015-10-14 16:48:34 +11:00 committed by Brant Knudson
parent 9959f63acf
commit 8f21658c00
8 changed files with 484 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
keystoneauth1/access/access.py
View File

@ -369,6 +369,22 @@ class AccessInfo(object):
"""
raise NotImplementedError()
@property
def bind(self):
"""Information about external mechanisms the token is bound to.
If a token is bound to an external authentication mechanism it can only
be used in conjunction with that mechanism. For example if bound to a
kerberos principal it may only be accepted if there is also kerberos
authentication performed on the request.
:returns: A dictionary or None. The key will be the bind type the value
is a dictionary that is specific to the format of the bind
type. Returns None if there is no bind information in the
token.
"""
raise NotImplementedError()
class AccessInfoV2(AccessInfo):
"""An object for encapsulating a raw v2 auth token from identity
@ -379,14 +395,14 @@ class AccessInfoV2(AccessInfo):
_service_catalog_class = service_catalog.ServiceCatalogV2
def has_service_catalog(self):
return 'serviceCatalog' in self
return 'serviceCatalog' in self._data.get('access', {})
@_missingproperty
def auth_token(self):
set_token = super(AccessInfoV2, self).auth_token
return set_token or self._data['access']['token']['id']
@_missingproperty
@property
def _token(self):
return self._data['access']['token']
@ -396,7 +412,7 @@ class AccessInfoV2(AccessInfo):
@_missingproperty
def issued(self):
return self._token['issued_at']
return utils.parse_isotime(self._token['issued_at'])
@property
def _user(self):
@ -420,7 +436,8 @@ class AccessInfoV2(AccessInfo):
@_missingproperty
def role_ids(self):
return self.get('metadata', {}).get('roles', [])
metadata = self._data.get('access', {}).get('metadata', {})
return metadata.get('roles', [])
@_missingproperty
def role_names(self):
@ -471,7 +488,7 @@ class AccessInfoV2(AccessInfo):
def trust_id(self):
return self._trust['id']
@property
@_missingproperty
def trust_scoped(self):
return bool(self._trust)
@ -543,6 +560,10 @@ class AccessInfoV2(AccessInfo):
def service_providers(self):
return None
@_missingproperty
def bind(self):
return self._token['bind']
class AccessInfoV3(AccessInfo):
"""An object for encapsulating a raw v3 auth token from identity
@ -707,3 +728,7 @@ class AccessInfoV3(AccessInfo):
service_providers.ServiceProviders.from_token(self._data))
return self._service_providers
@_missingproperty
def bind(self):
return self._data['token']['bind']

3
keystoneauth1/fixture/v2.py
View File

@ -242,3 +242,6 @@ class Token(dict):
def set_trust(self, id=None, trustee_user_id=None):
self.trust_id = id or uuid.uuid4().hex
self.trustee_user_id = trustee_user_id or uuid.uuid4().hex
def set_bind(self, name, data):
self._token.setdefault('bind', {})[name] = data

3
keystoneauth1/fixture/v3.py
View File

@ -405,6 +405,9 @@ class Token(dict):
_service_providers.append(sp)
return sp
def set_bind(self, name, data):
self.root.setdefault('bind', {})[name] = data
class V3FederationToken(Token):
"""A V3 Keystone Federation token that can be used for testing.

2
keystoneauth1/loading/opts.py
View File

@ -63,7 +63,7 @@ class Opt(object):
"you need to import it into your application's "
"requirements file. ")
deprecated_opts = [o._to_oslo_opt() for o in self.deprecated]
deprecated_opts = [cfg.DeprecatedOpt(o.name) for o in self.deprecated]
return cfg.Opt(name=self.name,
type=self.type,

212
keystoneauth1/tests/unit/access/test_v2_access.py Normal file
View File

@ -0,0 +1,212 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import datetime
import uuid
from oslo_utils import timeutils
from keystoneauth1 import access
from keystoneauth1 import fixture
from keystoneauth1.tests.unit import utils
class AccessV2Test(utils.TestCase):
def test_building_unscoped_accessinfo(self):
token = fixture.V2Token(expires='2012-10-03T16:58:01Z')
auth_ref = access.create(body=token)
self.assertIsInstance(auth_ref, access.AccessInfoV2)
self.assertFalse(auth_ref.has_service_catalog())
self.assertEqual(auth_ref.auth_token, token.token_id)
self.assertEqual(auth_ref.username, token.user_name)
self.assertEqual(auth_ref.user_id, token.user_id)
self.assertEqual(auth_ref.role_ids, [])
self.assertEqual(auth_ref.role_names, [])
self.assertIsNone(auth_ref.tenant_name)
self.assertIsNone(auth_ref.tenant_id)
self.assertFalse(auth_ref.domain_scoped)
self.assertFalse(auth_ref.project_scoped)
self.assertFalse(auth_ref.trust_scoped)
self.assertIsNone(auth_ref.project_domain_id)
self.assertIsNone(auth_ref.project_domain_name)
self.assertIsNone(auth_ref.user_domain_id)
self.assertIsNone(auth_ref.user_domain_name)
self.assertEqual(auth_ref.expires, token.expires)
self.assertEqual(auth_ref.issued, token.issued)
self.assertEqual(token.audit_id, auth_ref.audit_id)
self.assertIsNone(auth_ref.audit_chain_id)
self.assertIsNone(token.audit_chain_id)
self.assertIsNone(auth_ref.bind)
def test_will_expire_soon(self):
token = fixture.V2Token()
expires = timeutils.utcnow() + datetime.timedelta(minutes=5)
token.expires = expires
auth_ref = access.create(body=token)
self.assertIsInstance(auth_ref, access.AccessInfoV2)
self.assertFalse(auth_ref.will_expire_soon(stale_duration=120))
self.assertTrue(auth_ref.will_expire_soon(stale_duration=300))
self.assertFalse(auth_ref.will_expire_soon())
def test_building_scoped_accessinfo(self):
token = fixture.V2Token()
token.set_scope()
s = token.add_service('identity')
s.add_endpoint('http://url')
role_data = token.add_role()
auth_ref = access.create(body=token)
self.assertIsInstance(auth_ref, access.AccessInfoV2)
self.assertTrue(auth_ref.has_service_catalog())
self.assertEqual(auth_ref.auth_token, token.token_id)
self.assertEqual(auth_ref.username, token.user_name)
self.assertEqual(auth_ref.user_id, token.user_id)
self.assertEqual(auth_ref.role_ids, [role_data['id']])
self.assertEqual(auth_ref.role_names, [role_data['name']])
self.assertEqual(auth_ref.tenant_name, token.tenant_name)
self.assertEqual(auth_ref.tenant_id, token.tenant_id)
self.assertEqual(auth_ref.tenant_name, auth_ref.project_name)
self.assertEqual(auth_ref.tenant_id, auth_ref.project_id)
self.assertIsNone(auth_ref.project_domain_id, 'default')
self.assertIsNone(auth_ref.project_domain_name, 'Default')
self.assertIsNone(auth_ref.user_domain_id, 'default')
self.assertIsNone(auth_ref.user_domain_name, 'Default')
self.assertTrue(auth_ref.project_scoped)
self.assertFalse(auth_ref.domain_scoped)
self.assertEqual(token.audit_id, auth_ref.audit_id)
self.assertEqual(token.audit_chain_id, auth_ref.audit_chain_id)
def test_diablo_token(self):
diablo_token = {
'access': {
'token': {
'id': uuid.uuid4().hex,
'expires': '2020-01-01T00:00:10.000123Z',
'tenantId': 'tenant_id1',
},
'user': {
'id': 'user_id1',
'name': 'user_name1',
'roles': [
{'name': 'role1'},
{'name': 'role2'},
],
},
},
}
auth_ref = access.create(body=diablo_token)
self.assertIsInstance(auth_ref, access.AccessInfoV2)
self.assertTrue(auth_ref)
self.assertEqual(auth_ref.username, 'user_name1')
self.assertEqual(auth_ref.project_id, 'tenant_id1')
self.assertEqual(auth_ref.project_name, 'tenant_id1')
self.assertIsNone(auth_ref.project_domain_id)
self.assertIsNone(auth_ref.project_domain_name)
self.assertIsNone(auth_ref.user_domain_id)
self.assertIsNone(auth_ref.user_domain_name)
self.assertEqual(auth_ref.role_names, ['role1', 'role2'])
def test_grizzly_token(self):
grizzly_token = {
'access': {
'token': {
'id': uuid.uuid4().hex,
'expires': '2020-01-01T00:00:10.000123Z',
},
'user': {
'id': 'user_id1',
'name': 'user_name1',
'tenantId': 'tenant_id1',
'tenantName': 'tenant_name1',
'roles': [
{'name': 'role1'},
{'name': 'role2'},
],
},
},
}
auth_ref = access.create(body=grizzly_token)
self.assertIsInstance(auth_ref, access.AccessInfoV2)
self.assertEqual(auth_ref.project_id, 'tenant_id1')
self.assertEqual(auth_ref.project_name, 'tenant_name1')
self.assertIsNone(auth_ref.project_domain_id)
self.assertIsNone(auth_ref.project_domain_name)
self.assertIsNone(auth_ref.user_domain_id, 'default')
self.assertIsNone(auth_ref.user_domain_name, 'Default')
self.assertEqual(auth_ref.role_names, ['role1', 'role2'])
def test_v2_roles(self):
role_id = 'a'
role_name = 'b'
token = fixture.V2Token()
token.set_scope()
token.add_role(id=role_id, name=role_name)
auth_ref = access.create(body=token)
self.assertIsInstance(auth_ref, access.AccessInfoV2)
self.assertEqual([role_id], auth_ref.role_ids)
self.assertEqual([role_id],
auth_ref._data['access']['metadata']['roles'])
self.assertEqual([role_name], auth_ref.role_names)
self.assertEqual([{'name': role_name}],
auth_ref._data['access']['user']['roles'])
def test_trusts(self):
user_id = uuid.uuid4().hex
trust_id = uuid.uuid4().hex
token = fixture.V2Token(user_id=user_id, trust_id=trust_id)
token.set_scope()
token.add_role()
auth_ref = access.create(body=token)
self.assertIsInstance(auth_ref, access.AccessInfoV2)
self.assertEqual(trust_id, auth_ref.trust_id)
self.assertEqual(user_id, auth_ref.trustee_user_id)
self.assertEqual(trust_id, token['access']['trust']['id'])
def test_binding(self):
token = fixture.V2Token()
principal = uuid.uuid4().hex
token.set_bind('kerberos', principal)
auth_ref = access.create(body=token)
self.assertIsInstance(auth_ref, access.AccessInfoV2)
self.assertEqual({'kerberos': principal}, auth_ref.bind)

197
keystoneauth1/tests/unit/access/test_v3_access.py Normal file
View File

@ -0,0 +1,197 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import uuid
import datetime
from oslo_utils import timeutils
from keystoneauth1 import access
from keystoneauth1 import fixture
from keystoneauth1.tests.unit import utils
class AccessV3Test(utils.TestCase):
def test_building_unscoped_accessinfo(self):
token = fixture.V3Token()
token_id = uuid.uuid4().hex
auth_ref = access.create(body=token, auth_token=token_id)
self.assertIn('methods', auth_ref._data['token'])
self.assertFalse(auth_ref.has_service_catalog())
self.assertNotIn('catalog', auth_ref._data['token'])
self.assertEqual(token_id, auth_ref.auth_token)
self.assertEqual(token.user_name, auth_ref.username)
self.assertEqual(token.user_id, auth_ref.user_id)
self.assertEqual(auth_ref.role_ids, [])
self.assertEqual(auth_ref.role_names, [])
self.assertIsNone(auth_ref.project_name)
self.assertIsNone(auth_ref.project_id)
self.assertFalse(auth_ref.domain_scoped)
self.assertFalse(auth_ref.project_scoped)
self.assertEqual(token.user_domain_id, auth_ref.user_domain_id)
self.assertEqual(token.user_domain_name, auth_ref.user_domain_name)
self.assertIsNone(auth_ref.project_domain_id)
self.assertIsNone(auth_ref.project_domain_name)
self.assertEqual(auth_ref.expires, timeutils.parse_isotime(
token['token']['expires_at']))
self.assertEqual(auth_ref.issued, timeutils.parse_isotime(
token['token']['issued_at']))
self.assertEqual(auth_ref.expires, token.expires)
self.assertEqual(auth_ref.issued, token.issued)
self.assertEqual(auth_ref.audit_id, token.audit_id)
self.assertIsNone(auth_ref.audit_chain_id)
self.assertIsNone(token.audit_chain_id)
self.assertIsNone(auth_ref.bind)
def test_will_expire_soon(self):
expires = timeutils.utcnow() + datetime.timedelta(minutes=5)
token = fixture.V3Token(expires=expires)
auth_ref = access.create(body=token)
self.assertFalse(auth_ref.will_expire_soon(stale_duration=120))
self.assertTrue(auth_ref.will_expire_soon(stale_duration=301))
self.assertFalse(auth_ref.will_expire_soon())
def test_building_domain_scoped_accessinfo(self):
token = fixture.V3Token()
token.set_domain_scope()
s = token.add_service(type='identity')
s.add_standard_endpoints(public='http://url')
token_id = uuid.uuid4().hex
auth_ref = access.create(body=token, auth_token=token_id)
self.assertTrue(auth_ref)
self.assertIn('methods', auth_ref._data['token'])
self.assertIn('catalog', auth_ref._data['token'])
self.assertTrue(auth_ref.has_service_catalog())
self.assertTrue(auth_ref._data['token']['catalog'])
self.assertEqual(token_id, auth_ref.auth_token)
self.assertEqual(token.user_name, auth_ref.username)
self.assertEqual(token.user_id, auth_ref.user_id)
self.assertEqual(token.role_ids, auth_ref.role_ids)
self.assertEqual(token.role_names, auth_ref.role_names)
self.assertEqual(token.domain_name, auth_ref.domain_name)
self.assertEqual(token.domain_id, auth_ref.domain_id)
self.assertIsNone(auth_ref.project_name)
self.assertIsNone(auth_ref.project_id)
self.assertEqual(token.user_domain_id, auth_ref.user_domain_id)
self.assertEqual(token.user_domain_name, auth_ref.user_domain_name)
self.assertIsNone(auth_ref.project_domain_id)
self.assertIsNone(auth_ref.project_domain_name)
self.assertTrue(auth_ref.domain_scoped)
self.assertFalse(auth_ref.project_scoped)
self.assertEqual(token.audit_id, auth_ref.audit_id)
self.assertEqual(token.audit_chain_id, auth_ref.audit_chain_id)
def test_building_project_scoped_accessinfo(self):
token = fixture.V3Token()
token.set_project_scope()
s = token.add_service(type='identity')
s.add_standard_endpoints(public='http://url')
token_id = uuid.uuid4().hex
auth_ref = access.create(body=token, auth_token=token_id)
self.assertIn('methods', auth_ref._data['token'])
self.assertIn('catalog', auth_ref._data['token'])
self.assertTrue(auth_ref.has_service_catalog())
self.assertTrue(auth_ref._data['token']['catalog'])
self.assertEqual(token_id, auth_ref.auth_token)
self.assertEqual(token.user_name, auth_ref.username)
self.assertEqual(token.user_id, auth_ref.user_id)
self.assertEqual(token.role_ids, auth_ref.role_ids)
self.assertEqual(token.role_names, auth_ref.role_names)
self.assertIsNone(auth_ref.domain_name)
self.assertIsNone(auth_ref.domain_id)
self.assertEqual(token.project_name, auth_ref.project_name)
self.assertEqual(token.project_id, auth_ref.project_id)
self.assertEqual(auth_ref.tenant_name, auth_ref.project_name)
self.assertEqual(auth_ref.tenant_id, auth_ref.project_id)
self.assertEqual(token.project_domain_id, auth_ref.project_domain_id)
self.assertEqual(token.project_domain_name,
auth_ref.project_domain_name)
self.assertEqual(token.user_domain_id, auth_ref.user_domain_id)
self.assertEqual(token.user_domain_name, auth_ref.user_domain_name)
self.assertFalse(auth_ref.domain_scoped)
self.assertTrue(auth_ref.project_scoped)
self.assertEqual(token.audit_id, auth_ref.audit_id)
self.assertEqual(token.audit_chain_id, auth_ref.audit_chain_id)
def test_oauth_access(self):
consumer_id = uuid.uuid4().hex
access_token_id = uuid.uuid4().hex
token = fixture.V3Token()
token.set_project_scope()
token.set_oauth(access_token_id=access_token_id,
consumer_id=consumer_id)
auth_ref = access.create(body=token)
self.assertEqual(consumer_id, auth_ref.oauth_consumer_id)
self.assertEqual(access_token_id, auth_ref.oauth_access_token_id)
self.assertEqual(consumer_id,
auth_ref._data['token']['OS-OAUTH1']['consumer_id'])
self.assertEqual(
access_token_id,
auth_ref._data['token']['OS-OAUTH1']['access_token_id'])
def test_federated_property_standard_token(self):
"""Check if is_federated property returns expected value."""
token = fixture.V3Token()
token.set_project_scope()
auth_ref = access.create(body=token)
self.assertFalse(auth_ref.is_federated)
def test_binding(self):
token = fixture.V3Token()
principal = uuid.uuid4().hex
token.set_bind('kerberos', principal)
auth_ref = access.create(body=token)
self.assertIsInstance(auth_ref, access.AccessInfoV3)
self.assertEqual({'kerberos': principal}, auth_ref.bind)

12
keystoneauth1/tests/unit/loading/test_conf.py
View File

@ -201,3 +201,15 @@ class ConfTests(utils.TestCase):
plugin_names = set([o.name for o in plugin_opts])
self.assertEqual(plugin_names, loaded_names)
def test_register_cfg(self):
loading.register_auth_conf_options(self.conf_fixture.conf,
group=self.GROUP)
def test_common_conf_options(self):
opts = loading.get_auth_common_conf_options()
self.assertEqual(2, len(opts))
auth_type = [o for o in opts if o.name == 'auth_type'][0]
self.assertEqual(1, len(auth_type.deprecated_opts))
self.assertIsInstance(auth_type.deprecated_opts[0], cfg.DeprecatedOpt)

26
keystoneauth1/tests/unit/test_fixtures.py
View File

@ -109,6 +109,19 @@ class V2TokenTests(utils.TestCase):
self.assertEqual(region, service['region'])
self.assertEqual(endpoint_id, service['id'])
def test_token_bind(self):
name1 = uuid.uuid4().hex
data1 = uuid.uuid4().hex
name2 = uuid.uuid4().hex
data2 = {uuid.uuid4().hex: uuid.uuid4().hex}
token = fixture.V2Token()
token.set_bind(name1, data1)
token.set_bind(name2, data2)
self.assertEqual({name1: data1, name2: data2},
token['access']['token']['bind'])
class V3TokenTests(utils.TestCase):
@ -271,3 +284,16 @@ class V3TokenTests(utils.TestCase):
self.assertEqual(ref_service_providers, token.service_providers)
self.assertEqual(ref_service_providers,
token['token']['service_providers'])
def test_token_bind(self):
name1 = uuid.uuid4().hex
data1 = uuid.uuid4().hex
name2 = uuid.uuid4().hex
data2 = {uuid.uuid4().hex: uuid.uuid4().hex}
token = fixture.V3Token()
token.set_bind(name1, data1)
token.set_bind(name2, data2)
self.assertEqual({name1: data1, name2: data2},
token['token']['bind'])