openstack
/
keystone
Code Issues Proposed changes

Allow domain admins to list users in groups with v3 policy 

Domain admins (with a domain scoped token) could not list members of
groups in their domain or groups of a user in their domain.
This was due to 2 reasons: the v3 policy rule
'identity:list_groups_for_user' was not evaluating the users domain
and the identity controller method protections of 'list_users_in_group'
and 'list_groups_for_user' were not providing the required targets for
the rules.

Change-Id: Ibf8442a2ceefc2bb0941bd5e7beba6c252b2ab36
Closes-Bug: #1433402
Closes-Bug: #1458994
This commit is contained in:
Rudolf Vriend 2016-05-25 18:49:47 +02:00
parent 79952ffbd4
commit 9e7f24c235
3 changed files with 88 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
etc/policy.v3cloudsample.json
View File

@ -55,7 +55,7 @@
"admin_and_matching_group_domain_id": "rule:admin_required and domain_id:%(group.domain_id)s",
"identity:get_group": "rule:cloud_admin or rule:admin_and_matching_target_group_domain_id",
"identity:list_groups": "rule:cloud_admin or rule:admin_and_matching_domain_id",
"identity:list_groups_for_user": "rule:owner or rule:admin_and_matching_domain_id",
"identity:list_groups_for_user": "rule:owner or rule:admin_and_matching_target_user_domain_id",
"identity:create_group": "rule:cloud_admin or rule:admin_and_matching_group_domain_id",
"identity:update_group": "rule:cloud_admin or rule:admin_and_matching_target_group_domain_id",
"identity:delete_group": "rule:cloud_admin or rule:admin_and_matching_target_group_domain_id",

15
keystone/identity/controllers.py
View File

@ -211,6 +211,11 @@ class UserV3(controller.V3Controller):
ref['group'] = self.identity_api.get_group(group_id)
self.check_protection(context, prep_info, ref)
def _check_group_protection(self, context, prep_info, group_id):
ref = {}
ref['group'] = self.identity_api.get_group(group_id)
self.check_protection(context, prep_info, ref)
@controller.protected()
@validation.validated(schema.user_create, 'user')
def create_user(self, context, user):
@ -229,7 +234,8 @@ class UserV3(controller.V3Controller):
hints=hints)
return UserV3.wrap_collection(context, refs, hints=hints)
@controller.filterprotected('domain_id', 'enabled', 'name')
@controller.filterprotected('domain_id', 'enabled', 'name',
callback=_check_group_protection)
def list_users_in_group(self, context, filters, group_id):
hints = UserV3.build_driver_hints(context, filters)
refs = self.identity_api.list_users_in_group(group_id, hints=hints)
@ -299,6 +305,11 @@ class GroupV3(controller.V3Controller):
super(GroupV3, self).__init__()
self.get_member_from_driver = self.identity_api.get_group
def _check_user_protection(self, context, prep_info, user_id):
ref = {}
ref['user'] = self.identity_api.get_user(user_id)
self.check_protection(context, prep_info, ref)
@controller.protected()
@validation.validated(schema.group_create, 'group')
def create_group(self, context, group):
@ -317,7 +328,7 @@ class GroupV3(controller.V3Controller):
hints=hints)
return GroupV3.wrap_collection(context, refs, hints=hints)
@controller.filterprotected('name')
@controller.filterprotected('name', callback=_check_user_protection)
def list_groups_for_user(self, context, filters, user_id):
hints = GroupV3.build_driver_hints(context, filters)
refs = self.identity_api.list_groups_for_user(user_id, hints=hints)

74
keystone/tests/unit/test_v3_protection.py
View File

@ -576,6 +576,8 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase,
- just_a_user has a non-admin role on both domainA and the project.
- admin_domain has admin_project, and user cloud_admin_user, with an
'admin' role on admin_project.
- domainA has two groups (group1, group2), domainB has one group
(group3)
We test various api protection rules from the cloud sample policy
file to make sure the sample is valid and that we correctly enforce it.
@ -668,6 +670,15 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase,
user_id=self.just_a_user['id'],
project_id=self.project['id'])
self.group1 = unit.new_group_ref(domain_id=self.domainA['id'])
self.group1 = self.identity_api.create_group(self.group1)
self.group2 = unit.new_group_ref(domain_id=self.domainA['id'])
self.group2 = self.identity_api.create_group(self.group2)
self.group3 = unit.new_group_ref(domain_id=self.domainB['id'])
self.group3 = self.identity_api.create_group(self.group3)
def _stati(self, expected_status):
# Return the expected return codes for APIs with and without data
# with any specified status overriding the normal values
@ -696,6 +707,33 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase,
self.post('/users', auth=self.auth, body={'user': user_ref},
expected_status=status_created)
def _test_group_management(self, group, expected=None):
status_OK, status_created, status_no_data = self._stati(expected)
entity_url = '/groups/%s' % group['id']
list_url = '/groups?domain_id=%s' % group['domain_id']
users_url = '/groups/%s/users' % group['id']
group_member_url = '/groups/%s/users/%s' % (group['id'],
self.just_a_user['id'])
self.get(entity_url, auth=self.auth,
expected_status=status_OK)
self.get(list_url, auth=self.auth,
expected_status=status_OK)
self.put(group_member_url, auth=self.auth,
expected_status=status_no_data)
self.get(users_url, auth=self.auth,
expected_status=status_OK)
group_ref = unit.new_group_ref(domain_id=group['domain_id'])
self.post('/groups', auth=self.auth, body={'group': group_ref},
expected_status=status_created)
self.delete(group_member_url, auth=self.auth,
expected_status=status_no_data)
group = {'description': 'Updated'}
self.patch(entity_url, auth=self.auth, body={'group': group},
expected_status=status_OK)
self.delete(entity_url, auth=self.auth,
expected_status=status_no_data)
def _test_project_management(self, domain_id, expected=None):
status_OK, status_created, status_no_data = self._stati(expected)
entity_url = '/projects/%s' % self.project['id']
@ -920,6 +958,42 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase,
self._test_user_management(self.domainA['id'])
def test_group_management(self):
# First, authenticate with a user that does not have the domain
# admin role - shouldn't be able to do much.
self.auth = self.build_authentication_request(
user_id=self.just_a_user['id'],
password=self.just_a_user['password'],
domain_id=self.domainA['id'])
self._test_group_management(
self.group1, expected=exception.ForbiddenAction.code)
# ...but should be able to list groups of which they are a member
url = '/users/%s/groups' % self.just_a_user['id']
self.get(url, auth=self.auth)
# Now, authenticate with a user that does have the domain admin role
self.auth = self.build_authentication_request(
user_id=self.domain_admin_user['id'],
password=self.domain_admin_user['password'],
domain_id=self.domainA['id'])
self._test_group_management(self.group1)
self._test_group_management(self.group3,
expected=exception.ForbiddenAction.code)
def test_group_management_by_cloud_admin(self):
# Test groups management with a cloud admin. This user should
# be able to manage groups in any domain.
self.auth = self.build_authentication_request(
user_id=self.cloud_admin_user['id'],
password=self.cloud_admin_user['password'],
project_id=self.admin_project['id'])
self._test_group_management(self.group1)
self._test_group_management(self.group3)
def test_project_management(self):
# First, authenticate with a user that does not have the project
# admin role - shouldn't be able to do much.