# Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you 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. from marvin.cloudstackAPI import * from marvin.cloudstackTestCase import cloudstackTestCase from marvin.cloudstackException import CloudstackAPIException from marvin.lib.base import Account, Role, RolePermission, Configurations from marvin.lib.utils import cleanup_resources from nose.plugins.attrib import attr from random import shuffle import copy import random import re import time class TestData(object): """Test data object that is required to create resources """ def __init__(self): self.testdata = { "account": { "email": "mtu@test.cloud", "firstname": "Marvin", "lastname": "TestUser", "username": "roletest", "password": "password", }, "role": { "name": "MarvinFake Role ", "type": "User", "description": "Fake Role created by Marvin test" }, "importrole": { "name": "MarvinFake Import Role ", "type": "User", "description": "Fake Import User Role created by Marvin test", "rules" : [{"rule":"list*", "permission":"allow","description":"Listing apis"}, {"rule":"get*", "permission":"allow","description":"Get apis"}, {"rule":"update*", "permission":"deny","description":"Update apis"}] }, "roleadmin": { "name": "MarvinFake Admin Role ", "type": "Admin", "description": "Fake Admin Role created by Marvin test" }, "roledomainadmin": { "name": "MarvinFake DomainAdmin Role ", "type": "DomainAdmin", "description": "Fake Domain-Admin Role created by Marvin test" }, "rolepermission": { "roleid": 1, "rule": "listVirtualMachines", "permission": "allow", "description": "Fake role permission created by Marvin test" }, "apiConfig": { "listApis": "allow", "listAccounts": "allow", "listClusters": "deny", "*VM*": "allow", "*Host*": "deny" } } class TestDynamicRoles(cloudstackTestCase): """Tests dynamic role and role permission management in CloudStack """ def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.testdata = TestData().testdata feature_enabled = self.apiclient.listCapabilities(listCapabilities.listCapabilitiesCmd()).dynamicrolesenabled if not feature_enabled: self.skipTest("Dynamic Role-Based API checker not enabled, skipping test") self.testdata["role"]["name"] += self.getRandomString() self.role = Role.create( self.apiclient, self.testdata["role"] ) self.testdata["rolepermission"]["roleid"] = self.role.id self.rolepermission = RolePermission.create( self.apiclient, self.testdata["rolepermission"] ) self.account = Account.create( self.apiclient, self.testdata["account"], roleid=self.role.id ) cache_period_config = Configurations.list( self.apiclient, name='dynamic.apichecker.cache.period' )[0] self.cache_period = int(cache_period_config.value) self.cleanup = [ self.account, self.rolepermission, self.role ] def tearDown(self): try: cleanup_resources(self.apiclient, self.cleanup) except Exception as e: self.debug("Warning! Exception in tearDown: %s" % e) def translateRoleToAccountType(self, role_type): if role_type == "User": return 0 elif role_type == "Admin": return 1 elif role_type == "DomainAdmin": return 2 elif role_type == "ResourceAdmin": return 3 return -1 def getUserApiClient(self, username, domain='ROOT', role_type='User'): self.user_apiclient = self.testClient.getUserApiClient(UserName=username, DomainName='ROOT', type=self.translateRoleToAccountType(role_type)) return self.user_apiclient def getRandomString(self): return "".join(random.choice("abcdefghijklmnopqrstuvwxyz0123456789") for _ in range(10)) def getRandomRoleName(self): return "MarvinFakeRoleNewName-" + self.getRandomString() @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_lifecycle_list(self): """ Tests that default four roles exist """ roleTypes = {1: "Admin", 2: "ResourceAdmin", 3: "DomainAdmin", 4: "User"} for idx in range(1,5): list_roles = Role.list(self.apiclient, id=idx) self.assertEqual( isinstance(list_roles, list), True, "List Roles response was not a valid list" ) self.assertEqual( len(list_roles), 1, "List Roles response size was not 1" ) self.assertEqual( list_roles[0].type, roleTypes[idx], msg="Default role type differs from expectation" ) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_lifecycle_create(self): """ Tests normal lifecycle operations for roles """ # Reuse self.role created in setUp() try: role = Role.create( self.apiclient, self.testdata["role"] ) self.fail("An exception was expected when creating duplicate roles") except CloudstackAPIException: pass list_roles = Role.list(self.apiclient, id=self.role.id) self.assertEqual( isinstance(list_roles, list), True, "List Roles response was not a valid list" ) self.assertEqual( len(list_roles), 1, "List Roles response size was not 1" ) self.assertEqual( list_roles[0].name, self.testdata["role"]["name"], msg="Role name does not match the test data" ) self.assertEqual( list_roles[0].type, self.testdata["role"]["type"], msg="Role type does not match the test data" ) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_lifecycle_clone(self): """ Tests create role from existing role """ # Use self.role created in setUp() role_to_be_cloned = { "name": "MarvinFake Clone Role ", "roleid": self.role.id, "description": "Fake Role cloned by Marvin test" } try: role_cloned = Role.create( self.apiclient, role_to_be_cloned ) self.cleanup.append(role_cloned) except CloudstackAPIException as e: self.fail("Failed to create the role: %s" % e) list_role_cloned= Role.list(self.apiclient, id=role_cloned.id) self.assertEqual( isinstance(list_role_cloned, list), True, "List Roles response was not a valid list" ) self.assertEqual( len(list_role_cloned), 1, "List Roles response size was not 1" ) self.assertEqual( list_role_cloned[0].name, role_to_be_cloned["name"], msg="Role name does not match the test data" ) self.assertEqual( list_role_cloned[0].type, self.testdata["role"]["type"], msg="Role type does not match the test data" ) list_rolepermissions = RolePermission.list(self.apiclient, roleid=self.role.id) self.validate_permissions_list(list_rolepermissions, role_cloned.id) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_lifecycle_import(self): """ Tests import role with the rules """ # use importrole from testdata self.testdata["importrole"]["name"] += self.getRandomString() try: role_imported = Role.importRole( self.apiclient, self.testdata["importrole"] ) self.cleanup.append(role_imported) except CloudstackAPIException as e: self.fail("Failed to import the role: %s" % e) list_role_imported = Role.list(self.apiclient, id=role_imported.id) self.assertEqual( isinstance(list_role_imported, list), True, "List Roles response was not a valid list" ) self.assertEqual( len(list_role_imported), 1, "List Roles response size was not 1" ) self.assertEqual( list_role_imported[0].name, self.testdata["importrole"]["name"], msg="Role name does not match the test data" ) self.assertEqual( list_role_imported[0].type, self.testdata["importrole"]["type"], msg="Role type does not match the test data" ) self.validate_permissions_dict(self.testdata["importrole"]["rules"], role_imported.id) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_lifecycle_update(self): """ Tests role update """ self.account.delete(self.apiclient) new_role_name = self.getRandomRoleName() new_role_description = "Fake role description created after update" self.role.update(self.apiclient, name=new_role_name, type='Admin', description=new_role_description) update_role = Role.list(self.apiclient, id=self.role.id)[0] self.assertEqual( update_role.name, new_role_name, msg="Role name does not match updated role name" ) self.assertEqual( update_role.type, 'Admin', msg="Role type does not match updated role type" ) self.assertEqual( update_role.description, new_role_description, msg="Role description does not match updated role description" ) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_lifecycle_update_role_inuse(self): """ Tests role update when role is in use by an account """ new_role_name = self.getRandomRoleName() try: self.role.update(self.apiclient, name=new_role_name, type='Admin') self.fail("Updation of role type is not allowed when role is in use") except CloudstackAPIException: pass self.role.update(self.apiclient, name=new_role_name) update_role = Role.list(self.apiclient, id=self.role.id)[0] self.assertEqual( update_role.name, new_role_name, msg="Role name does not match updated role name" ) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_lifecycle_delete(self): """ Tests role update """ self.account.delete(self.apiclient) self.role.delete(self.apiclient) list_roles = Role.list(self.apiclient, id=self.role.id) self.assertEqual( list_roles, None, "List Roles response should be empty" ) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_inuse_deletion(self): """ Test to ensure role in use cannot be deleted """ try: self.role.delete(self.apiclient) self.fail("Role with any account should not be allowed to be deleted") except CloudstackAPIException: pass @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_default_role_deletion(self): """ Test to ensure 4 default roles cannot be deleted """ for idx in range(1,5): cmd = deleteRole.deleteRoleCmd() cmd.id = idx try: self.apiclient.deleteRole(cmd) self.fail("Default role got deleted with id: " + idx) except CloudstackAPIException: pass @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_rolepermission_lifecycle_list(self): """ Tests listing of default role's permission """ for idx in range(1,5): list_rolepermissions = RolePermission.list(self.apiclient, roleid=idx) self.assertEqual( isinstance(list_rolepermissions, list), True, "List rolepermissions response was not a valid list" ) self.assertTrue( len(list_rolepermissions) > 0, "List rolepermissions response was empty" ) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_rolepermission_lifecycle_create(self): """ Tests creation of role permission """ # Reuse self.rolepermission created in setUp() try: rolepermission = RolePermission.create( self.apiclient, self.testdata["rolepermission"] ) self.fail("An exception was expected when creating duplicate role permissions") except CloudstackAPIException: pass list_rolepermissions = RolePermission.list(self.apiclient, roleid=self.role.id) self.assertEqual( isinstance(list_rolepermissions, list), True, "List rolepermissions response was not a valid list" ) self.assertNotEqual( len(list_rolepermissions), 0, "List rolepermissions response was empty" ) self.assertEqual( list_rolepermissions[0].rule, self.testdata["rolepermission"]["rule"], msg="Role permission rule does not match the test data" ) self.assertEqual( list_rolepermissions[0].permission, self.testdata["rolepermission"]["permission"], msg="Role permission permission-type does not match the test data" ) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_rolepermission_lifecycle_update(self): """ Tests order updation of role permission """ permissions = [self.rolepermission] rules = ['list*', '*Vol*', 'listCapabilities'] for rule in rules: data = copy.deepcopy(self.testdata["rolepermission"]) data['rule'] = rule permission = RolePermission.create( self.apiclient, data ) self.cleanup.append(permission) permissions.append(permission) # Move last item to the top rule = permissions.pop(len(permissions)-1) permissions = [rule] + permissions rule.update(self.apiclient, ruleorder=",".join([x.id for x in permissions])) self.validate_permissions_list(permissions, self.role.id) # Move to the bottom rule = permissions.pop(0) permissions = permissions + [rule] rule.update(self.apiclient, ruleorder=",".join([x.id for x in permissions])) self.validate_permissions_list(permissions, self.role.id) # Random shuffles for _ in range(3): shuffle(permissions) rule.update(self.apiclient, ruleorder=",".join([x.id for x in permissions])) self.validate_permissions_list(permissions, self.role.id) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_rolepermission_lifecycle_update_permission(self): """ Tests update of Allow to Deny permission of a rule """ permissions = [self.rolepermission] rule = permissions.pop(0) rule.update(self.apiclient, ruleid=rule.id, permission='deny') list_rolepermissions = RolePermission.list(self.apiclient, roleid=self.role.id) self.assertEqual( list_rolepermissions[0].permission, 'deny', msg="List of role permissions do not match created list of permissions" ) rule.update(self.apiclient, ruleid=rule.id, permission='allow') list_rolepermissions = RolePermission.list(self.apiclient, roleid=self.role.id) self.assertEqual( list_rolepermissions[0].permission, 'allow', msg="List of role permissions do not match created list of permissions" ) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_rolepermission_lifecycle_update_permission_negative(self): """ Tests negative test for setting incorrect value as permission """ permissions = [self.rolepermission] rule = permissions.pop(0) try: rule.update(self.apiclient, ruleid=rule.id, permission='some_other_value') except Exception: pass else: self.fail("Negative test: Setting permission to 'some_other_value' should not be successful, failing") @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_rolepermission_lifecycle_concurrent_updates(self): """ Tests concurrent order updation of role permission """ permissions = [self.rolepermission] rules = ['list*', '*Vol*', 'listCapabilities'] for rule in rules: data = copy.deepcopy(self.testdata["rolepermission"]) data['rule'] = rule permission = RolePermission.create( self.apiclient, data ) self.cleanup.append(permission) permissions.append(permission) # The following rule is considered to be created by another mgmt server data = copy.deepcopy(self.testdata["rolepermission"]) data['rule'] = "someRule*" permission = RolePermission.create( self.apiclient, data ) self.cleanup.append(permission) shuffle(permissions) try: permission.update(self.apiclient, ruleorder=",".join([x.id for x in permissions])) self.fail("Reordering should fail in case of concurrent updates by other user") except CloudstackAPIException: pass @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_rolepermission_lifecycle_delete(self): """ Tests deletion of role permission """ permission = self.cleanup.pop(1) permission.delete(self.apiclient) list_rolepermissions = RolePermission.list(self.apiclient, roleid=self.role.id) self.assertEqual( list_rolepermissions, None, "List rolepermissions response should be empty" ) def checkApiAvailability(self, apiConfig, userApiClient): """ Checks available APIs based on api map """ response = userApiClient.listApis(listApis.listApisCmd()) allowedApis = [x.name for x in response] for api in allowedApis: for rule, perm in list(apiConfig.items()): if re.match(rule.replace('*', '.*'), api): if perm.lower() == 'allow': break else: self.fail('Denied API found to be allowed: ' + api) def checkApiCall(self, apiConfig, userApiClient): """ Performs actual API calls to verify API ACLs """ list_accounts = userApiClient.listAccounts(listAccounts.listAccountsCmd()) self.assertEqual( isinstance(list_accounts, list), True, "List accounts response was not a valid list" ) self.assertNotEqual( len(list_accounts), 0, "List accounts response was empty" ) # Perform actual API call for deny API try: userApiClient.listHosts(listHosts.listHostsCmd()) self.fail("API call succeeded which is denied for the role") except CloudstackAPIException: pass # Perform actual API call for API with no allow/deny rule try: userApiClient.listZones(listZones.listZonesCmd()) self.fail("API call succeeded which has no allow/deny rule for the role") except CloudstackAPIException: pass @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_account_acls(self): """ Test to check role, role permissions and account life cycles """ apiConfig = self.testdata['apiConfig'] for api, perm in list(apiConfig.items()): testdata = self.testdata['rolepermission'] testdata['roleid'] = self.role.id testdata['rule'] = api testdata['permission'] = perm.lower() RolePermission.create( self.apiclient, testdata ) time.sleep(self.cache_period + 5) userApiClient = self.getUserApiClient(self.account.name, domain=self.account.domain, role_type=self.account.roletype) # Perform listApis check self.checkApiAvailability(apiConfig, userApiClient) # Perform actual API call for allow API self.checkApiCall(apiConfig, userApiClient) @attr(tags=['advanced', 'simulator', 'basic', 'sg'], required_hardware=False) def test_role_account_acls_multiple_mgmt_servers(self): """ Test for role-rule enforcement in case of multiple mgmt servers Inserts rule directly in DB and checks expected behaviour """ apiConfig = self.testdata["apiConfig"] roleId = self.dbclient.execute("select id from roles where uuid='%s'" % self.role.id)[0][0] sortOrder = 1 for rule, perm in list(apiConfig.items()): self.dbclient.execute("insert into role_permissions (uuid, role_id, rule, permission, sort_order) values (UUID(), %d, '%s', '%s', %d)" % (roleId, rule, perm.upper(), sortOrder)) sortOrder += 1 time.sleep(self.cache_period + 5) userApiClient = self.getUserApiClient(self.account.name, domain=self.account.domain, role_type=self.account.roletype) # Perform listApis check self.checkApiAvailability(apiConfig, userApiClient) # Perform actual API call for allow API self.checkApiCall(apiConfig, userApiClient) def validate_permissions_list(self, permissions, roleid): list_rolepermissions = RolePermission.list(self.apiclient, roleid=roleid) self.assertEqual( len(list_rolepermissions), len(permissions), msg="List of role permissions do not match created list of permissions" ) for idx, rolepermission in enumerate(list_rolepermissions): self.assertEqual( rolepermission.rule, permissions[idx].rule, msg="Rule permission don't match with expected item at the index" ) self.assertEqual( rolepermission.permission, permissions[idx].permission, msg="Rule permission don't match with expected item at the index" ) def validate_permissions_dict(self, permissions, roleid): list_rolepermissions = RolePermission.list(self.apiclient, roleid=roleid) self.assertEqual( len(list_rolepermissions), len(permissions), msg="List of role permissions do not match created list of permissions" ) for idx, rolepermission in enumerate(list_rolepermissions): self.assertEqual( rolepermission.rule, permissions[idx]["rule"], msg="Rule permission don't match with expected item at the index" ) self.assertEqual( rolepermission.permission, permissions[idx]["permission"], msg="Rule permission don't match with expected item at the index" )