Browse Source

Merge "Add federated support for get user"

changes/02/719102/1
Zuul 1 year ago
committed by Gerrit Code Review
parent
commit
ffc235845a
  1. 4
      keystone/identity/core.py
  2. 40
      keystone/identity/shadow_backends/base.py
  3. 13
      keystone/identity/shadow_backends/sql.py
  4. 29
      keystone/tests/unit/test_shadow_users.py
  5. 10
      keystone/tests/unit/test_v3_identity.py

4
keystone/identity/core.py

@ -958,6 +958,10 @@ class Manager(manager.Manager):
domain_id, driver, entity_id = (
self._get_domain_driver_and_entity_id(user_id))
ref = driver.get_user(entity_id)
# Add user's federated objects
fed_objects = self.shadow_users_api.get_federated_objects(user_id)
if fed_objects:
ref['federated'] = fed_objects
return self._set_domain_id_and_mapping(
ref, domain_id, driver, mapping.EntityType.USER)

40
keystone/identity/shadow_backends/base.py

@ -17,6 +17,36 @@ import abc
from keystone import exception
def federated_objects_to_list(fed_ref):
"""Create a new reformatted federated object list using the one passed in.
When returning federated objects with a user we only need the attributes
idp_id, protocol_id, and unique_id. Therefore, we pull these elements out
of the fed_ref and create a newly formatted list with the needed
information. We simply group each federated object's protocol_ids and
unique_ids under the corresponding idp_id.
:returns list: Containing the user's federated objects
"""
if not fed_ref:
return []
fed = {}
for fed_dict in fed_ref:
fed.setdefault(
fed_dict['idp_id'],
{
'idp_id': fed_dict['idp_id'],
'protocols': []
}
)['protocols'].append({
'protocol_id': fed_dict['protocol_id'],
'unique_id': fed_dict['unique_id']
})
return list(fed.values())
class ShadowUsersDriverBase(object, metaclass=abc.ABCMeta):
"""Interface description for an Shadow Users driver."""
@ -32,6 +62,16 @@ class ShadowUsersDriverBase(object, metaclass=abc.ABCMeta):
"""
raise exception.NotImplemented()
@abc.abstractmethod
def get_federated_objects(self, user_id):
"""Get all federated objects for a user.
:param user_id: Unique identifier of the user
:returns list: Containing the user's federated objects
"""
raise exception.NotImplemented()
@abc.abstractmethod
def get_federated_user(self, idp_id, protocol_id, unique_id):
"""Return the found user for the federated identity.

13
keystone/identity/shadow_backends/sql.py

@ -54,6 +54,19 @@ class ShadowUsers(base.ShadowUsersDriverBase):
session.add(user_ref)
return identity_base.filter_user(user_ref.to_dict())
def get_federated_objects(self, user_id):
with sql.session_for_read() as session:
query = session.query(model.FederatedUser)
query = query.filter(model.FederatedUser.user_id == user_id)
fed_ref = []
for row in query:
m = model.FederatedUser(
idp_id=row.idp_id,
protocol_id=row.protocol_id,
unique_id=row.unique_id)
fed_ref.append(m.to_dict())
return base.federated_objects_to_list(fed_ref)
def _update_query_with_federated_statements(self, hints, query):
statements = []
for filter_ in hints.filters:

29
keystone/tests/unit/test_shadow_users.py

@ -59,3 +59,32 @@ class ShadowUsersTests(unit.TestCase,
self.idp['id'], self.protocol['id'], self.protocol)
self.domain_id = (
PROVIDERS.federation_api.get_idp(self.idp['id'])['domain_id'])
class TestUserWithFederatedUser(ShadowUsersTests):
def setUp(self):
super(TestUserWithFederatedUser, self).setUp()
self.useFixture(database.Database())
self.load_backends()
def assertFederatedDictsEqual(self, fed_dict, fed_object):
self.assertEqual(fed_dict['idp_id'], fed_object['idp_id'])
self.assertEqual(fed_dict['protocol_id'],
fed_object['protocols'][0]['protocol_id'])
self.assertEqual(fed_dict['unique_id'],
fed_object['protocols'][0]['unique_id'])
def test_get_user_when_user_has_federated_object(self):
fed_dict = unit.new_federated_user_ref(idp_id=self.idp['id'],
protocol_id=self.protocol['id'])
user = self.shadow_users_api.create_federated_user(
self.domain_id, fed_dict)
# test that the user returns a federated object and that there is only
# one returned
user_ref = self.identity_api.get_user(user['id'])
self.assertIn('federated', user_ref)
self.assertEqual(1, len(user_ref['federated']))
self.assertFederatedDictsEqual(fed_dict, user_ref['federated'][0])

10
keystone/tests/unit/test_v3_identity.py

@ -1167,3 +1167,13 @@ class UserFederatedAttributesTests(test_v3.RestfulTestCase):
def test_list_users_with_all_federated_attributes(self):
attribute = ['idp_id', 'protocol_id', 'unique_id']
self._test_list_users_with_federated_parameter(attribute)
def test_get_user_includes_required_federated_attributes(self):
user = self.identity_api.get_user(self.fed_user['id'])
self.assertIn('federated', user)
self.assertIn('idp_id', user['federated'][0])
self.assertIn('protocols', user['federated'][0])
self.assertIn('protocol_id', user['federated'][0]['protocols'][0])
self.assertIn('unique_id', user['federated'][0]['protocols'][0])
r = self.get('/users/%(user_id)s' % {'user_id': user['id']})
self.assertValidUserResponse(r, user)
Loading…
Cancel
Save