Add API route for list role assignments for tree
This patch adds the API routing (and appropriate REST API tests) to call the manager support for listing role assignments for a tree of projects that was implemented in the earlier patch. In order to support the required policy rule, the protection wrapper for filter list calls was extended to support a callback (in the same way that the regular protection wrapper already did). Implements: bp list-assignment-subtree Change-Id: I3495c7cab3b40811b2722ac7d70ddda30410b62b
This commit is contained in:
Henry Nash
parent
44d0c2f5a5
commit
a9a47b62c8
|
|
@ -79,6 +79,7 @@ identity:create_grant PUT `grant_resources`
|
|||
identity:revoke_grant DELETE `grant_resources`_
|
||||
|
||||
identity:list_role_assignments GET /v3/role_assignments
|
||||
identity:list_role_assignments_for_tree GET /v3/role_assignments?include_subtree
|
||||
|
||||
identity:get_policy GET /v3/policy/{policy_id}
|
||||
identity:list_policies GET /v3/policy
|
||||
|
|
|
|||
|
|
@ -82,6 +82,7 @@
|
|||
"identity:revoke_grant": "rule:admin_required",
|
||||
|
||||
"identity:list_role_assignments": "rule:admin_required",
|
||||
"identity:list_role_assignments_for_tree": "rule:admin_required",
|
||||
|
||||
"identity:get_policy": "rule:admin_required",
|
||||
"identity:list_policies": "rule:admin_required",
|
||||
|
|
|
|||
|
|
@ -92,8 +92,9 @@
|
|||
|
||||
"admin_on_domain_filter" : "rule:admin_required and domain_id:%(scope.domain.id)s",
|
||||
"admin_on_project_filter" : "rule:admin_required and project_id:%(scope.project.id)s",
|
||||
"admin_on_domain_of_project_filter" : "rule:admin_required and domain_id:%(target.project.domain_id)s",
|
||||
"identity:list_role_assignments": "rule:cloud_admin or rule:admin_on_domain_filter or rule:admin_on_project_filter",
|
||||
|
||||
"identity:list_role_assignments_for_tree": "rule:cloud_admin or rule:admin_on_domain_of_project_filter",
|
||||
"identity:get_policy": "rule:cloud_admin",
|
||||
"identity:list_policies": "rule:cloud_admin",
|
||||
"identity:create_policy": "rule:cloud_admin",
|
||||
|
|
|
|||
|
|
@ -590,10 +590,7 @@ class RoleAssignmentV3(controller.V3Controller):
|
|||
msg = _('Specify a user or group, not both')
|
||||
raise exception.ValidationError(msg)
|
||||
|
||||
@controller.filterprotected('group.id', 'role.id',
|
||||
'scope.domain.id', 'scope.project.id',
|
||||
'scope.OS-INHERIT:inherited_to', 'user.id')
|
||||
def list_role_assignments(self, context, filters):
|
||||
def _list_role_assignments(self, context, filters, include_subtree=False):
|
||||
"""List role assignments to user and groups on domains and projects.
|
||||
|
||||
Return a list of all existing role assignments in the system, filtered
|
||||
|
|
@ -644,8 +641,58 @@ class RoleAssignmentV3(controller.V3Controller):
|
|||
group_id=params.get('group.id'),
|
||||
domain_id=params.get('scope.domain.id'),
|
||||
project_id=params.get('scope.project.id'),
|
||||
include_subtree=include_subtree,
|
||||
inherited=inherited, effective=effective)
|
||||
|
||||
formatted_refs = [self._format_entity(context, ref) for ref in refs]
|
||||
|
||||
return self.wrap_collection(context, formatted_refs)
|
||||
|
||||
@controller.filterprotected('group.id', 'role.id',
|
||||
'scope.domain.id', 'scope.project.id',
|
||||
'scope.OS-INHERIT:inherited_to', 'user.id')
|
||||
def list_role_assignments(self, context, filters):
|
||||
return self._list_role_assignments(context, filters)
|
||||
|
||||
def _check_list_tree_protection(self, context, protection_info):
|
||||
"""Check protection for list assignment for tree API.
|
||||
|
||||
The policy rule might want to inspect the domain of any project filter
|
||||
so if one is defined, then load the project ref and pass it to the
|
||||
check protection method.
|
||||
|
||||
"""
|
||||
ref = {}
|
||||
for filter, value in protection_info['filter_attr'].items():
|
||||
if filter == 'scope.project.id' and value:
|
||||
ref['project'] = self.resource_api.get_project(value)
|
||||
|
||||
self.check_protection(context, protection_info, ref)
|
||||
|
||||
@controller.filterprotected('group.id', 'role.id',
|
||||
'scope.domain.id', 'scope.project.id',
|
||||
'scope.OS-INHERIT:inherited_to', 'user.id',
|
||||
callback=_check_list_tree_protection)
|
||||
def list_role_assignments_for_tree(self, context, filters):
|
||||
if not context['query_string'].get('scope.project.id'):
|
||||
msg = _('scope.project.id must be specified if include_subtree '
|
||||
'is also specified')
|
||||
raise exception.ValidationError(message=msg)
|
||||
return self._list_role_assignments(context, filters,
|
||||
include_subtree=True)
|
||||
|
||||
def list_role_assignments_wrapper(self, context):
|
||||
"""Main entry point from router for list role assignments.
|
||||
|
||||
Since we want different policy file rules to be applicable based on
|
||||
whether there the include_subtree query parameter is part of the API
|
||||
call, this method checks for this and then calls the appropriate
|
||||
protected entry point.
|
||||
|
||||
"""
|
||||
params = context['query_string']
|
||||
if 'include_subtree' in params and (
|
||||
self.query_filter_is_true(params['include_subtree'])):
|
||||
return self.list_role_assignments_for_tree(context)
|
||||
else:
|
||||
return self.list_role_assignments(context)
|
||||
|
|
|
|||
|
|
@ -162,7 +162,7 @@ class Routers(wsgi.RoutersBase):
|
|||
self._add_resource(
|
||||
mapper, controllers.RoleAssignmentV3(),
|
||||
path='/role_assignments',
|
||||
get_action='list_role_assignments',
|
||||
get_action='list_role_assignments_wrapper',
|
||||
rel=json_home.build_v3_resource_relation('role_assignments'))
|
||||
|
||||
if CONF.os_inherit.enabled:
|
||||
|
|
|
|||
|
|
@ -165,23 +165,32 @@ def protected(callback=None):
|
|||
return wrapper
|
||||
|
||||
|
||||
def filterprotected(*filters):
|
||||
"""Wraps filtered API calls with role based access controls (RBAC)."""
|
||||
def filterprotected(*filters, **callback):
|
||||
"""Wraps API list calls with role based access controls (RBAC).
|
||||
|
||||
This handles both the protection of the API parameters as well as any
|
||||
filters supplied.
|
||||
|
||||
More complex API list calls (for example that need to examine the contents
|
||||
of an entity referenced by one of the filters) should pass in a callback
|
||||
function, that will be subsequently called to check protection for these
|
||||
multiple entities. This callback function should gather the appropriate
|
||||
entities needed and then call check_protection() in the V3Controller class.
|
||||
|
||||
"""
|
||||
def _filterprotected(f):
|
||||
@functools.wraps(f)
|
||||
def wrapper(self, context, **kwargs):
|
||||
if not context['is_admin']:
|
||||
action = 'identity:%s' % f.__name__
|
||||
creds = _build_policy_check_credentials(self, action,
|
||||
context, kwargs)
|
||||
# Now, build the target dict for policy check. We include:
|
||||
# The target dict for the policy check will include:
|
||||
#
|
||||
# - Any query filter parameters
|
||||
# - Data from the main url (which will be in the kwargs
|
||||
# parameter) and would typically include the prime key
|
||||
# of a get/update/delete call
|
||||
# parameter), which although most of our APIs do not utilize,
|
||||
# in theory you could have.
|
||||
#
|
||||
# First any query filter parameters
|
||||
|
||||
# First build the dict of filter parameters
|
||||
target = dict()
|
||||
if filters:
|
||||
for item in filters:
|
||||
|
|
@ -192,15 +201,29 @@ def filterprotected(*filters):
|
|||
', '.join(['%s=%s' % (item, target[item])
|
||||
for item in target])))
|
||||
|
||||
# Now any formal url parameters
|
||||
for key in kwargs:
|
||||
target[key] = kwargs[key]
|
||||
if 'callback' in callback and callback['callback'] is not None:
|
||||
# A callback has been specified to load additional target
|
||||
# data, so pass it the formal url params as well as the
|
||||
# list of filters, so it can augment these and then call
|
||||
# the check_protection() method.
|
||||
prep_info = {'f_name': f.__name__,
|
||||
'input_attr': kwargs,
|
||||
'filter_attr': target}
|
||||
callback['callback'](self, context, prep_info, **kwargs)
|
||||
else:
|
||||
# No callback, so we are going to check the protection here
|
||||
action = 'identity:%s' % f.__name__
|
||||
creds = _build_policy_check_credentials(self, action,
|
||||
context, kwargs)
|
||||
# Add in any formal url parameters
|
||||
for key in kwargs:
|
||||
target[key] = kwargs[key]
|
||||
|
||||
self.policy_api.enforce(creds,
|
||||
action,
|
||||
utils.flatten_dict(target))
|
||||
self.policy_api.enforce(creds,
|
||||
action,
|
||||
utils.flatten_dict(target))
|
||||
|
||||
LOG.debug('RBAC: Authorization granted')
|
||||
LOG.debug('RBAC: Authorization granted')
|
||||
else:
|
||||
LOG.warning(_LW('RBAC: Bypassing authorization'))
|
||||
return f(self, context, filters, **kwargs)
|
||||
|
|
@ -781,6 +804,8 @@ class V3Controller(wsgi.Application):
|
|||
if target_attr:
|
||||
policy_dict = {'target': target_attr}
|
||||
policy_dict.update(prep_info['input_attr'])
|
||||
if 'filter_attr' in prep_info:
|
||||
policy_dict.update(prep_info['filter_attr'])
|
||||
self.policy_api.enforce(creds,
|
||||
action,
|
||||
utils.flatten_dict(policy_dict))
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ import uuid
|
|||
from oslo_config import cfg
|
||||
from six.moves import http_client
|
||||
from six.moves import range
|
||||
from testtools import matchers
|
||||
|
||||
from keystone.tests import unit
|
||||
from keystone.tests.unit import test_v3
|
||||
|
|
@ -1941,6 +1942,154 @@ class AssignmentInheritanceTestCase(test_v3.RestfulTestCase,
|
|||
inher_up_entity['scope']['project']['id'] = leaf_id
|
||||
self.assertRoleAssignmentInListResponse(r, inher_up_entity)
|
||||
|
||||
def test_project_id_specified_if_include_subtree_specified(self):
|
||||
"""When using include_subtree, you must specify a project ID."""
|
||||
self.get('/role_assignments?include_subtree=True',
|
||||
expected_status=http_client.BAD_REQUEST)
|
||||
self.get('/role_assignments?scope.project.id&'
|
||||
'include_subtree=True',
|
||||
expected_status=http_client.BAD_REQUEST)
|
||||
|
||||
def test_get_role_assignments_for_project_tree(self):
|
||||
"""Get role_assignment?scope.project.id=X?include_subtree``.
|
||||
|
||||
Test Plan:
|
||||
|
||||
- Create 2 roles and a hierarchy of projects with one root and one leaf
|
||||
- Issue the URL to add a non-inherited user role to the root project
|
||||
and the leaf project
|
||||
- Issue the URL to get role assignments for the root project but
|
||||
not the subtree - this should return just the root assignment
|
||||
- Issue the URL to get role assignments for the root project and
|
||||
it's subtree - this should return both assignments
|
||||
- Check that explicitly setting include_subtree to False is the
|
||||
equivalent to not including it at all in the query.
|
||||
|
||||
"""
|
||||
# Create default scenario
|
||||
root_id, leaf_id, non_inherited_role_id, unused_role_id = (
|
||||
self._setup_hierarchical_projects_scenario())
|
||||
|
||||
# Grant non-inherited role to root and leaf projects
|
||||
non_inher_entity_root = self.build_role_assignment_entity(
|
||||
project_id=root_id, user_id=self.user['id'],
|
||||
role_id=non_inherited_role_id)
|
||||
self.put(non_inher_entity_root['links']['assignment'])
|
||||
non_inher_entity_leaf = self.build_role_assignment_entity(
|
||||
project_id=leaf_id, user_id=self.user['id'],
|
||||
role_id=non_inherited_role_id)
|
||||
self.put(non_inher_entity_leaf['links']['assignment'])
|
||||
|
||||
# Without the subtree, we should get the one assignment on the
|
||||
# root project
|
||||
collection_url = (
|
||||
'/role_assignments?scope.project.id=%(project)s' % {
|
||||
'project': root_id})
|
||||
r = self.get(collection_url)
|
||||
self.assertValidRoleAssignmentListResponse(
|
||||
r, resource_url=collection_url)
|
||||
|
||||
self.assertThat(r.result['role_assignments'], matchers.HasLength(1))
|
||||
self.assertRoleAssignmentInListResponse(r, non_inher_entity_root)
|
||||
|
||||
# With the subtree, we should get both assignments
|
||||
collection_url = (
|
||||
'/role_assignments?scope.project.id=%(project)s'
|
||||
'&include_subtree=True' % {
|
||||
'project': root_id})
|
||||
r = self.get(collection_url)
|
||||
self.assertValidRoleAssignmentListResponse(
|
||||
r, resource_url=collection_url)
|
||||
|
||||
self.assertThat(r.result['role_assignments'], matchers.HasLength(2))
|
||||
self.assertRoleAssignmentInListResponse(r, non_inher_entity_root)
|
||||
self.assertRoleAssignmentInListResponse(r, non_inher_entity_leaf)
|
||||
|
||||
# With subtree=0, we should also only get the one assignment on the
|
||||
# root project
|
||||
collection_url = (
|
||||
'/role_assignments?scope.project.id=%(project)s'
|
||||
'&include_subtree=0' % {
|
||||
'project': root_id})
|
||||
r = self.get(collection_url)
|
||||
self.assertValidRoleAssignmentListResponse(
|
||||
r, resource_url=collection_url)
|
||||
|
||||
self.assertThat(r.result['role_assignments'], matchers.HasLength(1))
|
||||
self.assertRoleAssignmentInListResponse(r, non_inher_entity_root)
|
||||
|
||||
def test_get_effective_role_assignments_for_project_tree(self):
|
||||
"""Get role_assignment ?project_id=X?include_subtree=True?effective``.
|
||||
|
||||
Test Plan:
|
||||
|
||||
- Create 2 roles and a hierarchy of projects with one root and 4 levels
|
||||
of child project
|
||||
- Issue the URL to add a non-inherited user role to the root project
|
||||
and a level 1 project
|
||||
- Issue the URL to add an inherited user role on the level 2 project
|
||||
- Issue the URL to get effective role assignments for the level 1
|
||||
project and it's subtree - this should return a role (non-inherited)
|
||||
on the level 1 project and roles (inherited) on each of the level
|
||||
2, 3 and 4 projects
|
||||
|
||||
"""
|
||||
# Create default scenario
|
||||
root_id, leaf_id, non_inherited_role_id, inherited_role_id = (
|
||||
self._setup_hierarchical_projects_scenario())
|
||||
|
||||
# Add some extra projects to the project hierarchy
|
||||
level2 = unit.new_project_ref(domain_id=self.domain['id'],
|
||||
parent_id=leaf_id)
|
||||
level3 = unit.new_project_ref(domain_id=self.domain['id'],
|
||||
parent_id=level2['id'])
|
||||
level4 = unit.new_project_ref(domain_id=self.domain['id'],
|
||||
parent_id=level3['id'])
|
||||
self.resource_api.create_project(level2['id'], level2)
|
||||
self.resource_api.create_project(level3['id'], level3)
|
||||
self.resource_api.create_project(level4['id'], level4)
|
||||
|
||||
# Grant non-inherited role to root (as a spoiler) and to
|
||||
# the level 1 (leaf) project
|
||||
non_inher_entity_root = self.build_role_assignment_entity(
|
||||
project_id=root_id, user_id=self.user['id'],
|
||||
role_id=non_inherited_role_id)
|
||||
self.put(non_inher_entity_root['links']['assignment'])
|
||||
non_inher_entity_leaf = self.build_role_assignment_entity(
|
||||
project_id=leaf_id, user_id=self.user['id'],
|
||||
role_id=non_inherited_role_id)
|
||||
self.put(non_inher_entity_leaf['links']['assignment'])
|
||||
|
||||
# Grant inherited role to level 2
|
||||
inher_entity = self.build_role_assignment_entity(
|
||||
project_id=level2['id'], user_id=self.user['id'],
|
||||
role_id=inherited_role_id, inherited_to_projects=True)
|
||||
self.put(inher_entity['links']['assignment'])
|
||||
|
||||
# Get effective role assignments
|
||||
collection_url = (
|
||||
'/role_assignments?scope.project.id=%(project)s'
|
||||
'&include_subtree=True&effective' % {
|
||||
'project': leaf_id})
|
||||
r = self.get(collection_url)
|
||||
self.assertValidRoleAssignmentListResponse(
|
||||
r, resource_url=collection_url)
|
||||
|
||||
# There should be three assignments returned in total
|
||||
self.assertThat(r.result['role_assignments'], matchers.HasLength(3))
|
||||
|
||||
# Assert that the user does not non-inherited role on root project
|
||||
self.assertRoleAssignmentNotInListResponse(r, non_inher_entity_root)
|
||||
|
||||
# Assert that the user does have non-inherited role on leaf project
|
||||
self.assertRoleAssignmentInListResponse(r, non_inher_entity_leaf)
|
||||
|
||||
# Assert that the user has inherited role on levels 3 and 4
|
||||
inher_entity['scope']['project']['id'] = level3['id']
|
||||
self.assertRoleAssignmentInListResponse(r, inher_entity)
|
||||
inher_entity['scope']['project']['id'] = level4['id']
|
||||
self.assertRoleAssignmentInListResponse(r, inher_entity)
|
||||
|
||||
def test_get_inherited_role_assignments_for_project_hierarchy(self):
|
||||
"""Call ``GET /role_assignments?scope.OS-INHERIT:inherited_to``.
|
||||
|
||||
|
|
|
|||
|
|
@ -985,6 +985,53 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase,
|
|||
self.assertRoleAssignmentInListResponse(r, project_admin_entity)
|
||||
self.assertRoleAssignmentInListResponse(r, project_user_entity)
|
||||
|
||||
def test_domain_admin_list_assignment_tree(self):
|
||||
# Add a child project to the standard test data
|
||||
sub_project = unit.new_project_ref(domain_id=self.domainA['id'],
|
||||
parent_id=self.project['id'])
|
||||
self.resource_api.create_project(sub_project['id'], sub_project)
|
||||
self.assignment_api.create_grant(self.role['id'],
|
||||
user_id=self.just_a_user['id'],
|
||||
project_id=sub_project['id'])
|
||||
|
||||
collection_url = self.build_role_assignment_query_url(
|
||||
project_id=self.project['id'])
|
||||
collection_url += '&include_subtree=True'
|
||||
|
||||
# The domain admin should be able to list the assignment tree
|
||||
auth = self.build_authentication_request(
|
||||
user_id=self.domain_admin_user['id'],
|
||||
password=self.domain_admin_user['password'],
|
||||
domain_id=self.domainA['id'])
|
||||
|
||||
r = self.get(collection_url, auth=auth)
|
||||
self.assertValidRoleAssignmentListResponse(
|
||||
r, expected_length=3, resource_url=collection_url)
|
||||
|
||||
# A project admin should not be able to
|
||||
auth = self.build_authentication_request(
|
||||
user_id=self.project_admin_user['id'],
|
||||
password=self.project_admin_user['password'],
|
||||
project_id=self.project['id'])
|
||||
|
||||
r = self.get(collection_url, auth=auth,
|
||||
expected_status=http_client.FORBIDDEN)
|
||||
|
||||
# A neither should a domain admin from a different domain
|
||||
domainB_admin_user = unit.create_user(
|
||||
self.identity_api,
|
||||
domain_id=self.domainB['id'])
|
||||
self.assignment_api.create_grant(self.admin_role['id'],
|
||||
user_id=domainB_admin_user['id'],
|
||||
domain_id=self.domainB['id'])
|
||||
auth = self.build_authentication_request(
|
||||
user_id=domainB_admin_user['id'],
|
||||
password=domainB_admin_user['password'],
|
||||
domain_id=self.domainB['id'])
|
||||
|
||||
r = self.get(collection_url, auth=auth,
|
||||
expected_status=http_client.FORBIDDEN)
|
||||
|
||||
def test_domain_user_list_assignments_of_project_failed(self):
|
||||
self.auth = self.build_authentication_request(
|
||||
user_id=self.just_a_user['id'],
|
||||
|
|
|
|||
Loading…
Reference in New Issue