diff options
Diffstat (limited to 'keystone-moon/keystone/tests/unit/test_v3_protection.py')
| -rw-r--r-- | keystone-moon/keystone/tests/unit/test_v3_protection.py | 739 |
1 files changed, 610 insertions, 129 deletions
diff --git a/keystone-moon/keystone/tests/unit/test_v3_protection.py b/keystone-moon/keystone/tests/unit/test_v3_protection.py index 9922ae5e..f77a1528 100644 --- a/keystone-moon/keystone/tests/unit/test_v3_protection.py +++ b/keystone-moon/keystone/tests/unit/test_v3_protection.py @@ -20,19 +20,22 @@ from oslo_serialization import jsonutils from six.moves import http_client from keystone import exception -from keystone.policy.backends import rules from keystone.tests import unit +from keystone.tests.unit import ksfixtures from keystone.tests.unit.ksfixtures import temporaryfile from keystone.tests.unit import test_v3 +from keystone.tests.unit import utils CONF = cfg.CONF -DEFAULT_DOMAIN_ID = CONF.identity.default_domain_id class IdentityTestProtectedCase(test_v3.RestfulTestCase): """Test policy enforcement on the v3 Identity API.""" + def _policy_fixture(self): + return ksfixtures.Policy(self.tmpfilename, self.config_fixture) + def setUp(self): """Setup for Identity Protection Test Cases. @@ -49,14 +52,9 @@ class IdentityTestProtectedCase(test_v3.RestfulTestCase): the default domain. """ - # Ensure that test_v3.RestfulTestCase doesn't load its own - # sample data, which would make checking the results of our - # tests harder - super(IdentityTestProtectedCase, self).setUp() self.tempfile = self.useFixture(temporaryfile.SecureTempFile()) self.tmpfilename = self.tempfile.file_name - self.config_fixture.config(group='oslo_policy', - policy_file=self.tmpfilename) + super(IdentityTestProtectedCase, self).setUp() # A default auth request we can use - un-scoped user token self.auth = self.build_authentication_request( @@ -66,45 +64,33 @@ class IdentityTestProtectedCase(test_v3.RestfulTestCase): def load_sample_data(self): self._populate_default_domain() # Start by creating a couple of domains - self.domainA = self.new_domain_ref() + self.domainA = unit.new_domain_ref() self.resource_api.create_domain(self.domainA['id'], self.domainA) - self.domainB = self.new_domain_ref() + self.domainB = unit.new_domain_ref() self.resource_api.create_domain(self.domainB['id'], self.domainB) - self.domainC = self.new_domain_ref() - self.domainC['enabled'] = False + self.domainC = unit.new_domain_ref(enabled=False) self.resource_api.create_domain(self.domainC['id'], self.domainC) # Now create some users, one in domainA and two of them in domainB - self.user1 = self.new_user_ref(domain_id=self.domainA['id']) - password = uuid.uuid4().hex - self.user1['password'] = password - self.user1 = self.identity_api.create_user(self.user1) - self.user1['password'] = password - - self.user2 = self.new_user_ref(domain_id=self.domainB['id']) - password = uuid.uuid4().hex - self.user2['password'] = password - self.user2 = self.identity_api.create_user(self.user2) - self.user2['password'] = password - - self.user3 = self.new_user_ref(domain_id=self.domainB['id']) - password = uuid.uuid4().hex - self.user3['password'] = password - self.user3 = self.identity_api.create_user(self.user3) - self.user3['password'] = password - - self.group1 = self.new_group_ref(domain_id=self.domainA['id']) + self.user1 = unit.create_user(self.identity_api, + domain_id=self.domainA['id']) + self.user2 = unit.create_user(self.identity_api, + domain_id=self.domainB['id']) + self.user3 = unit.create_user(self.identity_api, + domain_id=self.domainB['id']) + + self.group1 = unit.new_group_ref(domain_id=self.domainA['id']) self.group1 = self.identity_api.create_group(self.group1) - self.group2 = self.new_group_ref(domain_id=self.domainA['id']) + self.group2 = unit.new_group_ref(domain_id=self.domainA['id']) self.group2 = self.identity_api.create_group(self.group2) - self.group3 = self.new_group_ref(domain_id=self.domainB['id']) + self.group3 = unit.new_group_ref(domain_id=self.domainB['id']) self.group3 = self.identity_api.create_group(self.group3) - self.role = self.new_role_ref() + self.role = unit.new_role_ref() self.role_api.create_role(self.role['id'], self.role) - self.role1 = self.new_role_ref() + self.role1 = unit.new_role_ref() self.role_api.create_role(self.role1['id'], self.role1) self.assignment_api.create_grant(self.role['id'], user_id=self.user1['id'], @@ -348,34 +334,23 @@ class IdentityTestPolicySample(test_v3.RestfulTestCase): def load_sample_data(self): self._populate_default_domain() - self.just_a_user = self.new_user_ref( + self.just_a_user = unit.create_user( + self.identity_api, domain_id=CONF.identity.default_domain_id) - password = uuid.uuid4().hex - self.just_a_user['password'] = password - self.just_a_user = self.identity_api.create_user(self.just_a_user) - self.just_a_user['password'] = password - - self.another_user = self.new_user_ref( + self.another_user = unit.create_user( + self.identity_api, domain_id=CONF.identity.default_domain_id) - password = uuid.uuid4().hex - self.another_user['password'] = password - self.another_user = self.identity_api.create_user(self.another_user) - self.another_user['password'] = password - - self.admin_user = self.new_user_ref( + self.admin_user = unit.create_user( + self.identity_api, domain_id=CONF.identity.default_domain_id) - password = uuid.uuid4().hex - self.admin_user['password'] = password - self.admin_user = self.identity_api.create_user(self.admin_user) - self.admin_user['password'] = password - self.role = self.new_role_ref() + self.role = unit.new_role_ref() self.role_api.create_role(self.role['id'], self.role) - self.admin_role = {'id': uuid.uuid4().hex, 'name': 'admin'} + self.admin_role = unit.new_role_ref(name='admin') self.role_api.create_role(self.admin_role['id'], self.admin_role) # Create and assign roles to the project - self.project = self.new_project_ref( + self.project = unit.new_project_ref( domain_id=CONF.identity.default_domain_id) self.resource_api.create_project(self.project['id'], self.project) self.assignment_api.create_grant(self.role['id'], @@ -461,7 +436,8 @@ class IdentityTestPolicySample(test_v3.RestfulTestCase): token = self.get_requested_token(auth) self.head('/auth/tokens', token=token, - headers={'X-Subject-Token': token}, expected_status=200) + headers={'X-Subject-Token': token}, + expected_status=http_client.OK) def test_user_check_user_token(self): # A user can check one of their own tokens. @@ -474,7 +450,8 @@ class IdentityTestPolicySample(test_v3.RestfulTestCase): token2 = self.get_requested_token(auth) self.head('/auth/tokens', token=token1, - headers={'X-Subject-Token': token2}, expected_status=200) + headers={'X-Subject-Token': token2}, + expected_status=http_client.OK) def test_user_check_other_user_token_rejected(self): # A user cannot check another user's token. @@ -510,7 +487,8 @@ class IdentityTestPolicySample(test_v3.RestfulTestCase): user_token = self.get_requested_token(user_auth) self.head('/auth/tokens', token=admin_token, - headers={'X-Subject-Token': user_token}, expected_status=200) + headers={'X-Subject-Token': user_token}, + expected_status=http_client.OK) def test_user_revoke_same_token(self): # Given a non-admin user token, the token can be used to revoke @@ -579,6 +557,10 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, test_v3.AssignmentTestMixin): """Test policy enforcement of the sample v3 cloud policy file.""" + def _policy_fixture(self): + return ksfixtures.Policy(unit.dirs.etc('policy.v3cloudsample.json'), + self.config_fixture) + def setUp(self): """Setup for v3 Cloud Policy Sample Test Cases. @@ -592,8 +574,8 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, - domain_admin_user has role 'admin' on domainA, - project_admin_user has role 'admin' on the project, - just_a_user has a non-admin role on both domainA and the project. - - admin_domain has user cloud_admin_user, with an 'admin' role - on admin_domain. + - admin_domain has admin_project, and user cloud_admin_user, with an + 'admin' role on admin_project. 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. @@ -604,62 +586,61 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, # tests harder super(IdentityTestv3CloudPolicySample, self).setUp() - # Finally, switch to the v3 sample policy file - self.addCleanup(rules.reset) - rules.reset() self.config_fixture.config( - group='oslo_policy', - policy_file=unit.dirs.etc('policy.v3cloudsample.json')) + group='resource', + admin_project_name=self.admin_project['name']) + self.config_fixture.config( + group='resource', + admin_project_domain_name=self.admin_domain['name']) def load_sample_data(self): # Start by creating a couple of domains self._populate_default_domain() - self.domainA = self.new_domain_ref() + self.domainA = unit.new_domain_ref() self.resource_api.create_domain(self.domainA['id'], self.domainA) - self.domainB = self.new_domain_ref() + self.domainB = unit.new_domain_ref() self.resource_api.create_domain(self.domainB['id'], self.domainB) - self.admin_domain = {'id': 'admin_domain_id', 'name': 'Admin_domain'} + self.admin_domain = unit.new_domain_ref() self.resource_api.create_domain(self.admin_domain['id'], self.admin_domain) + self.admin_project = unit.new_project_ref( + domain_id=self.admin_domain['id']) + self.resource_api.create_project(self.admin_project['id'], + self.admin_project) + # And our users - self.cloud_admin_user = self.new_user_ref( + self.cloud_admin_user = unit.create_user( + self.identity_api, domain_id=self.admin_domain['id']) - password = uuid.uuid4().hex - self.cloud_admin_user['password'] = password - self.cloud_admin_user = ( - self.identity_api.create_user(self.cloud_admin_user)) - self.cloud_admin_user['password'] = password - self.just_a_user = self.new_user_ref(domain_id=self.domainA['id']) - password = uuid.uuid4().hex - self.just_a_user['password'] = password - self.just_a_user = self.identity_api.create_user(self.just_a_user) - self.just_a_user['password'] = password - self.domain_admin_user = self.new_user_ref( + self.just_a_user = unit.create_user( + self.identity_api, domain_id=self.domainA['id']) - password = uuid.uuid4().hex - self.domain_admin_user['password'] = password - self.domain_admin_user = ( - self.identity_api.create_user(self.domain_admin_user)) - self.domain_admin_user['password'] = password - self.project_admin_user = self.new_user_ref( + self.domain_admin_user = unit.create_user( + self.identity_api, + domain_id=self.domainA['id']) + self.domainB_admin_user = unit.create_user( + self.identity_api, + domain_id=self.domainB['id']) + self.project_admin_user = unit.create_user( + self.identity_api, domain_id=self.domainA['id']) - password = uuid.uuid4().hex - self.project_admin_user['password'] = password - self.project_admin_user = ( - self.identity_api.create_user(self.project_admin_user)) - self.project_admin_user['password'] = password - - # The admin role and another plain role - self.admin_role = {'id': uuid.uuid4().hex, 'name': 'admin'} + self.project_adminB_user = unit.create_user( + self.identity_api, + domain_id=self.domainB['id']) + + # The admin role, a domain specific role and another plain role + self.admin_role = unit.new_role_ref(name='admin') self.role_api.create_role(self.admin_role['id'], self.admin_role) - self.role = self.new_role_ref() + self.roleA = unit.new_role_ref(domain_id=self.domainA['id']) + self.role_api.create_role(self.roleA['id'], self.roleA) + self.role = unit.new_role_ref() self.role_api.create_role(self.role['id'], self.role) - # The cloud admin just gets the admin role + # The cloud admin just gets the admin role on the special admin project self.assignment_api.create_grant(self.admin_role['id'], user_id=self.cloud_admin_user['id'], - domain_id=self.admin_domain['id']) + project_id=self.admin_project['id']) # Assign roles to the domain self.assignment_api.create_grant(self.admin_role['id'], @@ -668,13 +649,21 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.assignment_api.create_grant(self.role['id'], user_id=self.just_a_user['id'], domain_id=self.domainA['id']) + self.assignment_api.create_grant(self.admin_role['id'], + user_id=self.domainB_admin_user['id'], + domain_id=self.domainB['id']) # Create and assign roles to the project - self.project = self.new_project_ref(domain_id=self.domainA['id']) + self.project = unit.new_project_ref(domain_id=self.domainA['id']) self.resource_api.create_project(self.project['id'], self.project) + self.projectB = unit.new_project_ref(domain_id=self.domainB['id']) + self.resource_api.create_project(self.projectB['id'], self.projectB) self.assignment_api.create_grant(self.admin_role['id'], user_id=self.project_admin_user['id'], project_id=self.project['id']) + self.assignment_api.create_grant( + self.admin_role['id'], user_id=self.project_adminB_user['id'], + project_id=self.projectB['id']) self.assignment_api.create_grant(self.role['id'], user_id=self.just_a_user['id'], project_id=self.project['id']) @@ -683,7 +672,8 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, # Return the expected return codes for APIs with and without data # with any specified status overriding the normal values if expected_status is None: - return (200, 201, 204) + return (http_client.OK, http_client.CREATED, + http_client.NO_CONTENT) else: return (expected_status, expected_status, expected_status) @@ -702,7 +692,7 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.delete(entity_url, auth=self.auth, expected_status=status_no_data) - user_ref = self.new_user_ref(domain_id=domain_id) + user_ref = unit.new_user_ref(domain_id=domain_id) self.post('/users', auth=self.auth, body={'user': user_ref}, expected_status=status_created) @@ -721,7 +711,7 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.delete(entity_url, auth=self.auth, expected_status=status_no_data) - proj_ref = self.new_project_ref(domain_id=domain_id) + proj_ref = unit.new_project_ref(domain_id=domain_id) self.post('/projects', auth=self.auth, body={'project': proj_ref}, expected_status=status_created) @@ -740,13 +730,14 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.delete(entity_url, auth=self.auth, expected_status=status_no_data) - domain_ref = self.new_domain_ref() + domain_ref = unit.new_domain_ref() self.post('/domains', auth=self.auth, body={'domain': domain_ref}, expected_status=status_created) - def _test_grants(self, target, entity_id, expected=None): + def _test_grants(self, target, entity_id, role_domain_id=None, + list_status_OK=False, expected=None): status_OK, status_created, status_no_data = self._stati(expected) - a_role = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex} + a_role = unit.new_role_ref(domain_id=role_domain_id) self.role_api.create_role(a_role['id'], a_role) collection_url = ( @@ -762,11 +753,67 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, expected_status=status_no_data) self.head(member_url, auth=self.auth, expected_status=status_no_data) - self.get(collection_url, auth=self.auth, - expected_status=status_OK) + if list_status_OK: + self.get(collection_url, auth=self.auth) + else: + self.get(collection_url, auth=self.auth, + expected_status=status_OK) self.delete(member_url, auth=self.auth, expected_status=status_no_data) + def _role_management_cases(self, read_status_OK=False, expected=None): + # Set the different status values for different types of call depending + # on whether we expect the calls to fail or not. + status_OK, status_created, status_no_data = self._stati(expected) + entity_url = '/roles/%s' % self.role['id'] + list_url = '/roles' + + if read_status_OK: + self.get(entity_url, auth=self.auth) + self.get(list_url, auth=self.auth) + else: + self.get(entity_url, auth=self.auth, + expected_status=status_OK) + self.get(list_url, auth=self.auth, + expected_status=status_OK) + + role = {'name': 'Updated'} + self.patch(entity_url, auth=self.auth, body={'role': role}, + expected_status=status_OK) + self.delete(entity_url, auth=self.auth, + expected_status=status_no_data) + + role_ref = unit.new_role_ref() + self.post('/roles', auth=self.auth, body={'role': role_ref}, + expected_status=status_created) + + def _domain_role_management_cases(self, domain_id, read_status_OK=False, + expected=None): + # Set the different status values for different types of call depending + # on whether we expect the calls to fail or not. + status_OK, status_created, status_no_data = self._stati(expected) + entity_url = '/roles/%s' % self.roleA['id'] + list_url = '/roles?domain_id=%s' % domain_id + + if read_status_OK: + self.get(entity_url, auth=self.auth) + self.get(list_url, auth=self.auth) + else: + self.get(entity_url, auth=self.auth, + expected_status=status_OK) + self.get(list_url, auth=self.auth, + expected_status=status_OK) + + role = {'name': 'Updated'} + self.patch(entity_url, auth=self.auth, body={'role': role}, + expected_status=status_OK) + self.delete(entity_url, auth=self.auth, + expected_status=status_no_data) + + role_ref = unit.new_role_ref(domain_id=domain_id) + self.post('/roles', auth=self.auth, body={'role': role_ref}, + expected_status=status_created) + def test_user_management(self): # First, authenticate with a user that does not have the domain # admin role - shouldn't be able to do much. @@ -786,13 +833,90 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self._test_user_management(self.domainA['id']) + def test_user_management_normalized_keys(self): + """Illustrate the inconsistent handling of hyphens in keys. + + To quote Morgan in bug 1526244: + + the reason this is converted from "domain-id" to "domain_id" is + because of how we process/normalize data. The way we have to handle + specific data types for known columns requires avoiding "-" in the + actual python code since "-" is not valid for attributes in python + w/o significant use of "getattr" etc. + + In short, historically we handle some things in conversions. The + use of "extras" has long been a poor design choice that leads to + odd/strange inconsistent behaviors because of other choices made in + handling data from within the body. (In many cases we convert from + "-" to "_" throughout openstack) + + Source: https://bugs.launchpad.net/keystone/+bug/1526244/comments/9 + + """ + # Authenticate with a user that has 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']) + + # Show that we can read a normal user without any surprises. + r = self.get( + '/users/%s' % self.just_a_user['id'], + auth=self.auth, + expected_status=http_client.OK) + self.assertValidUserResponse(r) + + # We don't normalize query string keys, so both of these result in a + # 403, because we didn't specify a domain_id query string in either + # case, and we explicitly require one (it doesn't matter what + # 'domain-id' value you use). + self.get( + '/users?domain-id=%s' % self.domainA['id'], + auth=self.auth, + expected_status=exception.ForbiddenAction.code) + self.get( + '/users?domain-id=%s' % self.domainB['id'], + auth=self.auth, + expected_status=exception.ForbiddenAction.code) + + # If we try updating the user's 'domain_id' by specifying a + # 'domain-id', then it'll be stored into extras rather than normalized, + # and the user's actual 'domain_id' is not affected. + r = self.patch( + '/users/%s' % self.just_a_user['id'], + auth=self.auth, + body={'user': {'domain-id': self.domainB['id']}}, + expected_status=http_client.OK) + self.assertEqual(self.domainB['id'], r.json['user']['domain-id']) + self.assertEqual(self.domainA['id'], r.json['user']['domain_id']) + self.assertNotEqual(self.domainB['id'], self.just_a_user['domain_id']) + self.assertValidUserResponse(r, self.just_a_user) + + # Finally, show that we can create a new user without any surprises. + # But if we specify a 'domain-id' instead of a 'domain_id', we get a + # Forbidden response because we fail a policy check before + # normalization occurs. + user_ref = unit.new_user_ref(domain_id=self.domainA['id']) + r = self.post( + '/users', + auth=self.auth, + body={'user': user_ref}, + expected_status=http_client.CREATED) + self.assertValidUserResponse(r, ref=user_ref) + user_ref['domain-id'] = user_ref.pop('domain_id') + self.post( + '/users', + auth=self.auth, + body={'user': user_ref}, + expected_status=exception.ForbiddenAction.code) + def test_user_management_by_cloud_admin(self): # Test users management with a cloud admin. This user should # be able to manage users in any domain. self.auth = self.build_authentication_request( user_id=self.cloud_admin_user['id'], password=self.cloud_admin_user['password'], - domain_id=self.admin_domain['id']) + project_id=self.admin_project['id']) self._test_user_management(self.domainA['id']) @@ -824,7 +948,7 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.auth = self.build_authentication_request( user_id=self.cloud_admin_user['id'], password=self.cloud_admin_user['password'], - domain_id=self.admin_domain['id']) + project_id=self.admin_project['id']) # Check whether cloud admin can operate a domain # other than its own domain or not @@ -858,10 +982,56 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.auth = self.build_authentication_request( user_id=self.cloud_admin_user['id'], password=self.cloud_admin_user['password'], - domain_id=self.admin_domain['id']) + project_id=self.admin_project['id']) self._test_grants('domains', self.domainA['id']) + def test_domain_grants_by_cloud_admin_for_domain_specific_role(self): + # Test domain grants with a cloud admin. This user should be + # able to manage domain roles on 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_grants('domains', self.domainA['id'], + role_domain_id=self.domainB['id']) + + def test_domain_grants_by_non_admin_for_domain_specific_role(self): + # A non-admin shouldn't be able to do anything + 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_grants('domains', self.domainA['id'], + role_domain_id=self.domainA['id'], + expected=exception.ForbiddenAction.code) + self._test_grants('domains', self.domainA['id'], + role_domain_id=self.domainB['id'], + expected=exception.ForbiddenAction.code) + + def test_domain_grants_by_domain_admin_for_domain_specific_role(self): + # Authenticate with a user that does have the domain admin role, + # should not be able to assign a domain_specific role from another + # domain + 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_grants('domains', self.domainA['id'], + role_domain_id=self.domainB['id'], + # List status will always be OK, since we are not + # granting/checking/deleting assignments + list_status_OK=True, + expected=exception.ForbiddenAction.code) + + # They should be able to assign a domain specific role from the same + # domain + self._test_grants('domains', self.domainA['id'], + role_domain_id=self.domainA['id']) + def test_project_grants(self): self.auth = self.build_authentication_request( user_id=self.just_a_user['id'], @@ -890,11 +1060,67 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self._test_grants('projects', self.project['id']) + def test_project_grants_by_non_admin_for_domain_specific_role(self): + # A non-admin shouldn't be able to do anything + self.auth = self.build_authentication_request( + user_id=self.just_a_user['id'], + password=self.just_a_user['password'], + project_id=self.project['id']) + + self._test_grants('projects', self.project['id'], + role_domain_id=self.domainA['id'], + expected=exception.ForbiddenAction.code) + self._test_grants('projects', self.project['id'], + role_domain_id=self.domainB['id'], + expected=exception.ForbiddenAction.code) + + def test_project_grants_by_project_admin_for_domain_specific_role(self): + # Authenticate with a user that does have the project admin role, + # should not be able to assign a domain_specific role from another + # domain + self.auth = self.build_authentication_request( + user_id=self.project_admin_user['id'], + password=self.project_admin_user['password'], + project_id=self.project['id']) + + self._test_grants('projects', self.project['id'], + role_domain_id=self.domainB['id'], + # List status will always be OK, since we are not + # granting/checking/deleting assignments + list_status_OK=True, + expected=exception.ForbiddenAction.code) + + # They should be able to assign a domain specific role from the same + # domain + self._test_grants('projects', self.project['id'], + role_domain_id=self.domainA['id']) + + def test_project_grants_by_domain_admin_for_domain_specific_role(self): + # Authenticate with a user that does have the domain admin role, + # should not be able to assign a domain_specific role from another + # domain + 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_grants('projects', self.project['id'], + role_domain_id=self.domainB['id'], + # List status will always be OK, since we are not + # granting/checking/deleting assignments + list_status_OK=True, + expected=exception.ForbiddenAction.code) + + # They should be able to assign a domain specific role from the same + # domain + self._test_grants('projects', self.project['id'], + role_domain_id=self.domainA['id']) + def test_cloud_admin_list_assignments_of_domain(self): self.auth = self.build_authentication_request( user_id=self.cloud_admin_user['id'], password=self.cloud_admin_user['password'], - domain_id=self.admin_domain['id']) + project_id=self.admin_project['id']) collection_url = self.build_role_assignment_query_url( domain_id=self.domainA['id']) @@ -968,7 +1194,7 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.auth = self.build_authentication_request( user_id=self.cloud_admin_user['id'], password=self.cloud_admin_user['password'], - domain_id=self.admin_domain['id']) + project_id=self.admin_project['id']) collection_url = self.build_role_assignment_query_url( project_id=self.project['id']) @@ -990,7 +1216,33 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.assertRoleAssignmentInListResponse(r, project_admin_entity) self.assertRoleAssignmentInListResponse(r, project_user_entity) - @unit.utils.wip('waiting on bug #1437407') + def test_admin_project_list_assignments_of_project(self): + self.auth = self.build_authentication_request( + user_id=self.project_admin_user['id'], + password=self.project_admin_user['password'], + project_id=self.project['id']) + + collection_url = self.build_role_assignment_query_url( + project_id=self.project['id']) + r = self.get(collection_url, auth=self.auth) + self.assertValidRoleAssignmentListResponse( + r, expected_length=2, resource_url=collection_url) + + project_admin_entity = self.build_role_assignment_entity( + project_id=self.project['id'], + user_id=self.project_admin_user['id'], + role_id=self.admin_role['id'], + inherited_to_projects=False) + project_user_entity = self.build_role_assignment_entity( + project_id=self.project['id'], + user_id=self.just_a_user['id'], + role_id=self.role['id'], + inherited_to_projects=False) + + self.assertRoleAssignmentInListResponse(r, project_admin_entity) + self.assertRoleAssignmentInListResponse(r, project_user_entity) + + @utils.wip('waiting on bug #1437407') def test_domain_admin_list_assignments_of_project(self): self.auth = self.build_authentication_request( user_id=self.domain_admin_user['id'], @@ -1017,6 +1269,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'], @@ -1040,7 +1339,23 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.auth = self.build_authentication_request( user_id=self.cloud_admin_user['id'], password=self.cloud_admin_user['password'], - domain_id=self.admin_domain['id']) + project_id=self.admin_project['id']) + + self._test_domain_management() + + def test_admin_project(self): + self.auth = self.build_authentication_request( + user_id=self.project_admin_user['id'], + password=self.project_admin_user['password'], + project_id=self.project['id']) + + self._test_domain_management( + expected=exception.ForbiddenAction.code) + + 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_domain_management() @@ -1050,16 +1365,15 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, password=self.domain_admin_user['password'], domain_id=self.domainA['id']) entity_url = '/domains/%s' % self.domainA['id'] - self.get(entity_url, auth=self.auth, expected_status=200) + self.get(entity_url, auth=self.auth) def test_list_user_credentials(self): - self.credential_user = self.new_credential_ref(self.just_a_user['id']) - self.credential_api.create_credential(self.credential_user['id'], - self.credential_user) - self.credential_admin = self.new_credential_ref( - self.cloud_admin_user['id']) - self.credential_api.create_credential(self.credential_admin['id'], - self.credential_admin) + credential_user = unit.new_credential_ref(self.just_a_user['id']) + self.credential_api.create_credential(credential_user['id'], + credential_user) + credential_admin = unit.new_credential_ref(self.cloud_admin_user['id']) + self.credential_api.create_credential(credential_admin['id'], + credential_admin) self.auth = self.build_authentication_request( user_id=self.just_a_user['id'], @@ -1075,9 +1389,8 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, def test_get_and_delete_ec2_credentials(self): """Tests getting and deleting ec2 credentials through the ec2 API.""" - another_user = self.new_user_ref(domain_id=self.domainA['id']) - password = another_user['password'] - another_user = self.identity_api.create_user(another_user) + another_user = unit.create_user(self.identity_api, + domain_id=self.domainA['id']) # create a credential for just_a_user just_user_auth = self.build_authentication_request( @@ -1091,7 +1404,7 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, # another normal user can't get the credential another_user_auth = self.build_authentication_request( user_id=another_user['id'], - password=password) + password=another_user['password']) another_user_url = '/users/%s/credentials/OS-EC2/%s' % ( another_user['id'], r.result['credential']['access']) self.get(another_user_url, auth=another_user_auth, @@ -1160,7 +1473,26 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, admin_auth = self.build_authentication_request( user_id=self.cloud_admin_user['id'], password=self.cloud_admin_user['password'], - domain_id=self.admin_domain['id']) + project_id=self.admin_project['id']) + admin_token = self.get_requested_token(admin_auth) + + user_auth = self.build_authentication_request( + user_id=self.just_a_user['id'], + password=self.just_a_user['password']) + user_token = self.get_requested_token(user_auth) + + self.get('/auth/tokens', token=admin_token, + headers={'X-Subject-Token': user_token}) + + def test_admin_project_validate_user_token(self): + # An admin can validate a user's token. + # This is GET /v3/auth/tokens + + admin_auth = self.build_authentication_request( + user_id=self.project_admin_user['id'], + password=self.project_admin_user['password'], + project_id=self.project['id']) + admin_token = self.get_requested_token(admin_auth) user_auth = self.build_authentication_request( @@ -1182,7 +1514,8 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, token = self.get_requested_token(auth) self.head('/auth/tokens', token=token, - headers={'X-Subject-Token': token}, expected_status=200) + headers={'X-Subject-Token': token}, + expected_status=http_client.OK) def test_user_check_user_token(self): # A user can check one of their own tokens. @@ -1195,7 +1528,8 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, token2 = self.get_requested_token(auth) self.head('/auth/tokens', token=token1, - headers={'X-Subject-Token': token2}, expected_status=200) + headers={'X-Subject-Token': token2}, + expected_status=http_client.OK) def test_user_check_other_user_token_rejected(self): # A user cannot check another user's token. @@ -1231,7 +1565,8 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, user_token = self.get_requested_token(user_auth) self.head('/auth/tokens', token=admin_token, - headers={'X-Subject-Token': user_token}, expected_status=200) + headers={'X-Subject-Token': user_token}, + expected_status=http_client.OK) def test_user_revoke_same_token(self): # Given a non-admin user token, the token can be used to revoke @@ -1294,3 +1629,149 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase, self.delete('/auth/tokens', token=admin_token, headers={'X-Subject-Token': user_token}) + + def test_user_with_a_role_get_project(self): + user_auth = self.build_authentication_request( + user_id=self.just_a_user['id'], + password=self.just_a_user['password'], + project_id=self.project['id']) + + # Test user can get project for one they have a role in + self.get('/projects/%s' % self.project['id'], auth=user_auth) + + # Test user can not get project for one they don't have a role in, + # even if they have a role on another project + project2 = unit.new_project_ref(domain_id=self.domainA['id']) + self.resource_api.create_project(project2['id'], project2) + self.get('/projects/%s' % project2['id'], auth=user_auth, + expected_status=exception.ForbiddenAction.code) + + def test_project_admin_get_project(self): + admin_auth = self.build_authentication_request( + user_id=self.project_admin_user['id'], + password=self.project_admin_user['password'], + project_id=self.project['id']) + + resp = self.get('/projects/%s' % self.project['id'], auth=admin_auth) + self.assertEqual(self.project['id'], + jsonutils.loads(resp.body)['project']['id']) + + def test_role_management_no_admin_no_rights(self): + # A non-admin domain user shouldn't be able to manipulate roles + 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._role_management_cases(expected=exception.ForbiddenAction.code) + + # ...and nor should non-admin project user + self.auth = self.build_authentication_request( + user_id=self.just_a_user['id'], + password=self.just_a_user['password'], + project_id=self.project['id']) + + self._role_management_cases(expected=exception.ForbiddenAction.code) + + def test_role_management_with_project_admin(self): + # A project admin user should be able to get and list, but not be able + # to create/update/delete global roles + self.auth = self.build_authentication_request( + user_id=self.project_admin_user['id'], + password=self.project_admin_user['password'], + project_id=self.project['id']) + + self._role_management_cases(read_status_OK=True, + expected=exception.ForbiddenAction.code) + + def test_role_management_with_domain_admin(self): + # A domain admin user should be able to get and list, but not be able + # to create/update/delete global roles + 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._role_management_cases(read_status_OK=True, + expected=exception.ForbiddenAction.code) + + def test_role_management_with_cloud_admin(self): + # A cloud admin user should have rights to manipulate global roles + 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._role_management_cases() + + def test_domain_role_management_no_admin_no_rights(self): + # A non-admin domain user shouldn't be able to manipulate domain roles + 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._domain_role_management_cases( + self.domainA['id'], expected=exception.ForbiddenAction.code) + + # ...and nor should non-admin project user + self.auth = self.build_authentication_request( + user_id=self.just_a_user['id'], + password=self.just_a_user['password'], + project_id=self.project['id']) + + self._domain_role_management_cases( + self.domainA['id'], expected=exception.ForbiddenAction.code) + + def test_domain_role_management_with_cloud_admin(self): + # A cloud admin user should have rights to manipulate domain roles + 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._domain_role_management_cases(self.domainA['id']) + + def test_domain_role_management_with_domain_admin(self): + # A domain admin user should only be able to manipulate the domain + # specific roles in their own domain + self.auth = self.build_authentication_request( + user_id=self.domainB_admin_user['id'], + password=self.domainB_admin_user['password'], + domain_id=self.domainB['id']) + + # Try to access the domain specific roles in another domain + self._domain_role_management_cases( + self.domainA['id'], expected=exception.ForbiddenAction.code) + + # ...but they should be able to work with those in their own domain + 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._domain_role_management_cases(self.domainA['id']) + + def test_domain_role_management_with_project_admin(self): + # A project admin user should have not access to domain specific roles + # in another domain. They should be able to get and list domain + # specific roles from their own domain, but not be able to create, + # update or delete them, + self.auth = self.build_authentication_request( + user_id=self.project_adminB_user['id'], + password=self.project_adminB_user['password'], + project_id=self.projectB['id']) + + # Try access the domain specific roless in another domain + self._domain_role_management_cases( + self.domainA['id'], expected=exception.ForbiddenAction.code) + + # ...but they should be ablet to work with those in their own domain + self.auth = self.build_authentication_request( + user_id=self.project_admin_user['id'], + password=self.project_admin_user['password'], + project_id=self.project['id']) + + self._domain_role_management_cases( + self.domainA['id'], read_status_OK=True, + expected=exception.ForbiddenAction.code) |