| # 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. | |
| #Import Local Modules | |
| from .ldap_test_data import LdapTestData | |
| from marvin.cloudstackTestCase import cloudstackTestCase | |
| from marvin.lib.utils import (cleanup_resources) | |
| from marvin.lib.base import (listLdapUsers, | |
| ldapCreateAccount, | |
| importLdapUsers, | |
| User, | |
| Domain, | |
| Account, | |
| addLdapConfiguration, | |
| deleteLdapConfiguration, | |
| linkAccountToLdap, | |
| linkDomainToLdap, | |
| updateConfiguration) | |
| from marvin.lib.common import (get_domain, | |
| get_zone) | |
| from nose.plugins.attrib import attr | |
| # for login validation | |
| import requests | |
| import logging | |
| class TestLDAP(cloudstackTestCase): | |
| @classmethod | |
| def setUpClass(cls): | |
| ''' | |
| needs to | |
| - create the applicable ldap accounts in the directory server | |
| - create three domains: | |
| -- LDAP/manual | |
| -- LDAP/import | |
| -- LDAP/sync | |
| ''' | |
| cls.logger = logging.getLogger(__name__) | |
| stream_handler = logging.StreamHandler() | |
| logger_formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s') | |
| stream_handler.setFormatter(logger_formatter) | |
| cls.logger.setLevel(logging.DEBUG) | |
| cls.logger.addHandler(stream_handler) | |
| cls.logger.info("Setting up Class") | |
| testClient = super(TestLDAP, cls).getClsTestClient() | |
| cls.apiclient = testClient.getApiClient() | |
| try: | |
| # Setup test data | |
| cls.testdata = LdapTestData() | |
| if cls.config.TestData and cls.config.TestData.Path: | |
| cls.logger.debug("reading extra config from '" + cls.config.TestData.Path + "'") | |
| cls.testdata.update(cls.config.TestData.Path) | |
| cls.logger.debug(cls.testdata) | |
| cls.services = testClient.getParsedTestDataConfig() | |
| cls.services["configurableData"]["ldap_configuration"] = cls.testdata.testdata["ldap_configuration"] | |
| cls.logger.debug(cls.services["configurableData"]["ldap_configuration"]) | |
| # Get Zone, Domain | |
| cls.domain = get_domain(cls.apiclient) | |
| cls.logger.debug("standard domain: %s" % cls.domain.id) | |
| cls.zone = get_zone(cls.apiclient, cls.testClient.getZoneForTests()) | |
| cls._cleanup = [] | |
| # Build the test env | |
| cls.create_domains(cls.testdata) | |
| cls.configure_ldap_for_domains(cls.testdata) | |
| cls.test_user = [ | |
| cls.testdata.testdata[LdapTestData.users][0][LdapTestData.uid], | |
| cls.testdata.testdata[LdapTestData.users][1][LdapTestData.uid], | |
| cls.testdata.testdata[LdapTestData.users][2][LdapTestData.uid] | |
| ] | |
| except Exception as e: | |
| cls.logger.debug("Exception in setUpClass(cls): %s" % e) | |
| cls.tearDownClass() | |
| raise Exception("setup failed due to %s", e) | |
| return | |
| @classmethod | |
| def tearDownClass(cls): | |
| cls.logger.info("Tearing Down Class") | |
| try: | |
| super(TestLDAP, cls).tearDownClass() | |
| finally: | |
| try: | |
| cls.remove_ldap_configuration_for_domains() | |
| cls.logger.debug("done cleaning up resources in tearDownClass(cls) %s") | |
| except Exception as e: | |
| cls.logger.debug("Exception in tearDownClass(cls): %s" % e) | |
| def setUp(self): | |
| self.cleanup = [] | |
| self.server_details = self.config.__dict__["mgtSvr"][0].__dict__ | |
| self.server_url = "http://%s:8080/client/api" % self.server_details['mgtSvrIp'] | |
| return | |
| def tearDown(self): | |
| super(TestLDAP, self).tearDown() | |
| @attr(tags=["smoke", "advanced"], required_hardware="false") | |
| def test_01_manual(self): | |
| ''' | |
| test if an account can be imported | |
| prerequisite | |
| a ldap host is configured | |
| a domain is linked to cloudstack | |
| ''' | |
| cmd = listLdapUsers.listLdapUsersCmd() | |
| cmd.domainid = self.manualDomain.id | |
| cmd.userfilter = "LocalDomain" | |
| response = self.apiclient.listLdapUsers(cmd) | |
| self.logger.info("users found for linked domain %s" % response) | |
| self.assertEqual(len(response), len(self.testdata.testdata[LdapTestData.users]), "unexpected number (%d) of ldap users" % len(self.testdata.testdata[LdapTestData.users])) | |
| cmd = ldapCreateAccount.ldapCreateAccountCmd() | |
| cmd.domainid = self.manualDomain.id | |
| cmd.accounttype = 0 | |
| cmd.username = self.test_user[1] | |
| create_response = self.apiclient.ldapCreateAccount(cmd) | |
| # cleanup | |
| # last results id should be the account | |
| list_response = Account.list(self.apiclient, id=create_response.id) | |
| account_created = Account(list_response[0].__dict__) | |
| self.cleanup.append(account_created) | |
| self.assertEqual(len(create_response.user), 1, "only one user %s should be present" % self.test_user[1]) | |
| self.assertEqual(len(list_response), | |
| 1, | |
| "only one account (for user %s) should be present" % self.test_user[1]) | |
| return | |
| @attr(tags=["smoke", "advanced"], required_hardware="false") | |
| def test_02_import(self): | |
| ''' | |
| test if components are synced | |
| prerequisite | |
| a ldap host is configured | |
| a domain is linked to cloudstack | |
| ''' | |
| domainid = self.importDomain.id | |
| cmd = importLdapUsers.importLdapUsersCmd() | |
| cmd.domainid = domainid | |
| cmd.accounttype = 0 | |
| import_response = self.apiclient.importLdapUsers(cmd) | |
| # this is needed purely for cleanup: | |
| # cleanup | |
| list_response = Account.list(self.apiclient, domainid=domainid) | |
| for account in list_response: | |
| account_created = Account(account.__dict__) | |
| self.logger.debug("account to clean: %s (id: %s)" % (account_created.name, account_created.id)) | |
| self.cleanup.append(account_created) | |
| self.assertEqual(len(import_response), len(self.testdata.testdata[LdapTestData.users]), "unexpected number of ldap users") | |
| self.assertEqual(len(list_response), len(self.testdata.testdata[LdapTestData.users]), "only one account (for user %s) should be present" % self.test_user[1]) | |
| return | |
| @attr(tags=["smoke", "advanced"], required_hardware="false") | |
| def test_03_sync(self): | |
| ''' | |
| test if components are synced | |
| prerequisite | |
| a ldap host is configured | |
| a domain is linked to cloudstack | |
| some accounts in that domain are linked to groups in ldap | |
| ''' | |
| domainid = self.syncDomain.id | |
| username = self.test_user[1] | |
| # validate the user doesn't exist | |
| response = User.list(self.apiclient,domainid=domainid,username=username) | |
| self.assertEqual(response, None, "user should not exist yet") | |
| self.logon_test_user(username) | |
| # now validate the user exists in domain | |
| response = User.list(self.apiclient,domainid=domainid,username=username) | |
| for user in response: | |
| user_created = User(user.__dict__) | |
| self.logger.debug("user to clean: %s (id: %s)" % (user_created.username, user_created.id)) | |
| self.cleanup.append(user_created) | |
| # now verify the creation of the user | |
| self.assertEqual(len(response), 1, "user should exist by now") | |
| return | |
| @attr(tags=["smoke", "advanced"], required_hardware="false") | |
| def test_04_filtered_list_of_users(self): | |
| ''' | |
| test if we can get a filtered list of ldap users | |
| prerequisite | |
| a ldap host is configured | |
| a couple of ldapdomains are linked to cloudstack domains | |
| some accounts in those domain are linked to groups in ldap | |
| some ldap accounts are linked and present with the same uid | |
| some ldap accounts are not yet linked but present at other locations in cloudstack | |
| NOTE 1: if this test is run last only the explicitly imported test user from test_03_sync | |
| is in the system. The accounts from test_01_manual and test_02_import should have been cleared | |
| by the test tearDown(). We can not depend on test_03_sync having run so the test must avoid | |
| depending on it either being available or not. | |
| NOTE 2: this test will not work if the ldap users UIDs are already present in the ACS instance | |
| against which is being tested | |
| ''' | |
| cmd = listLdapUsers.listLdapUsersCmd() | |
| cmd.userfilter = "NoFilter" | |
| cmd.domainid = self.manualDomain.id | |
| response = self.apiclient.listLdapUsers(cmd) | |
| self.logger.debug(cmd.userfilter + " : " + str(response)) | |
| self.assertEqual(len(response), len(self.testdata.testdata[LdapTestData.users]), "unexpected number of ldap users") | |
| # create a non ldap user with the uid of cls.test_user[0] in parentDomain | |
| # create a manual import of a cls.test_user[1] in manualDomain | |
| # log on with test_user[2] in an syncDomain | |
| # we can now test all four filtertypes in syncDomain and inspect the respective outcomes for validity | |
| self.logon_test_user(self.test_user[2]) | |
| cmd.userfilter = "LocalDomain" | |
| cmd.domainid = self.syncDomain.id | |
| response = self.apiclient.listLdapUsers(cmd) | |
| self.logger.debug(cmd.userfilter + " : " + str(response)) | |
| self.assertEqual(len(response), | |
| len(self.testdata.testdata[LdapTestData.users]) - 1, | |
| "unexpected number of ldap users") | |
| @attr(tags=["smoke", "advanced"], required_hardware="false") | |
| def test_05_relink_account_and_reuse_user(self): | |
| ''' | |
| test if an account and thus a user can be removed and re-added | |
| test if components still are synced | |
| prerequisite | |
| a ldap host is configured | |
| a domain is linked to cloudstack | |
| some accounts in that domain are linked to groups in ldap | |
| ''' | |
| domainid = self.syncDomain.id | |
| username = self.test_user[1] | |
| # validate the user doesn't exist | |
| response = User.list(self.apiclient,domainid=domainid,username=username) | |
| self.assertEqual(response, None, "user should not exist yet") | |
| self.logon_test_user(username) | |
| # now validate the user exists in domain | |
| response = User.list(self.apiclient,domainid=domainid,username=username) | |
| # for user in response: | |
| # user_created = User(user.__dict__) | |
| # self.debug("user to clean: %s (id: %s)" % (user_created.username, user_created.id)) | |
| # # we don't cleanup to test if re-adding fails | |
| # self.cleanup.append(user_created) | |
| # now verify the creation of the user | |
| self.assertEqual(len(response), 1, "user should exist by now") | |
| # delete the account - quick implementation: user[1] happens to be a junior | |
| self.junior_account.delete(self.apiclient) | |
| # add the account with the same ldap group | |
| self.bind_account_to_ldap( | |
| account=self.testdata.testdata[LdapTestData.syncAccounts][0]["name"], | |
| ldapdomain=self.testdata.testdata[LdapTestData.syncAccounts][0]["group"], | |
| accounttype=self.testdata.testdata[LdapTestData.syncAccounts][0]["accounttype"]) | |
| # logon the user - should succeed - reported to fail | |
| self.logon_test_user(username) | |
| # now verify the creation of the user | |
| response = User.list(self.apiclient,domainid=domainid,username=username) | |
| for user in response: | |
| user_created = User(user.__dict__) | |
| self.debug("user to clean: %s (id: %s)" % (user_created.username, user_created.id)) | |
| # we don't cleanup to test if re-adding fails | |
| # self.cleanup.append(user_created) | |
| self.assertEqual(len(response), 1, "user should exist again") | |
| return | |
| def logon_test_user(self, username, domain = None): | |
| # login of dahn should create a user in account juniors | |
| args = {} | |
| args["command"] = 'login' | |
| args["username"] = username | |
| args["password"] = 'password' | |
| if domain == None: | |
| args["domain"] = "/" + self.parentDomain.name + "/" + self.syncDomain.name | |
| else: | |
| args["domain"] = domain | |
| args["response"] = "json" | |
| session = requests.Session() | |
| try: | |
| resp = session.post(self.server_url, params=args, verify=False) | |
| except requests.exceptions.ConnectionError as e: | |
| self.fail("Failed to attempt login request to mgmt server") | |
| @classmethod | |
| def create_domains(cls, td): | |
| # create a parent domain | |
| cls.parentDomain = cls.create_domain(td.testdata["domains"][0], parent_domain=cls.domain.id) | |
| cls.manualDomain = cls.create_domain(td.testdata["domains"][1], parent_domain=cls.parentDomain.id) | |
| cls.importDomain = cls.create_domain(td.testdata["domains"][2], parent_domain=cls.parentDomain.id) | |
| cls.syncDomain = cls.create_domain(td.testdata["domains"][3], parent_domain=cls.parentDomain.id) | |
| @classmethod | |
| def create_domain(cls, domain_to_create, parent_domain = None): | |
| cls.logger.debug("Creating domain: %s under %s" % (domain_to_create[LdapTestData.name], parent_domain)) | |
| if parent_domain: | |
| domain_to_create["parentdomainid"] = parent_domain | |
| tmpDomain = Domain.create(cls.apiclient, domain_to_create) | |
| cls._cleanup.append(tmpDomain) | |
| cls.logger.debug("Created domain %s with id %s " % (tmpDomain.name, tmpDomain.id)) | |
| return tmpDomain | |
| @classmethod | |
| def configure_ldap_for_domains(cls, td) : | |
| cmd = addLdapConfiguration.addLdapConfigurationCmd() | |
| cmd.hostname = td.testdata[LdapTestData.configuration][LdapTestData.hostname] | |
| cmd.port = td.testdata[LdapTestData.configuration][LdapTestData.port] | |
| cls.logger.debug("configuring ldap server for domain %s" % LdapTestData.manualDomain) | |
| cmd.domainid = cls.manualDomain.id | |
| response = cls.apiclient.addLdapConfiguration(cmd) | |
| cls.manualLdap = response | |
| cls.logger.debug("configuring ldap server for domain %s" % LdapTestData.importDomain) | |
| cmd.domainid = cls.importDomain.id | |
| response = cls.apiclient.addLdapConfiguration(cmd) | |
| cls.importLdap = response | |
| cls.logger.debug("configuring ldap server for domain %s" % LdapTestData.syncDomain) | |
| cmd.domainid = cls.syncDomain.id | |
| response = cls.apiclient.addLdapConfiguration(cmd) | |
| cls.syncLdap = response | |
| cls.set_ldap_settings_on_domain(domainid=cls.manualDomain.id) | |
| cls.bind_domain_to_ldap(domainid=cls.manualDomain.id, ldapdomain=cls.testdata.admins) | |
| cls.set_ldap_settings_on_domain(domainid=cls.importDomain.id) | |
| cls.bind_domain_to_ldap(domainid=cls.importDomain.id, ldapdomain=cls.testdata.admins, | |
| accounttype=2, type="Group") # just to be testing different types | |
| cls.set_ldap_settings_on_domain(domainid=cls.syncDomain.id) | |
| cls.create_sync_accounts() | |
| @classmethod | |
| def remove_ldap_configuration_for_domains(cls) : | |
| cls.logger.debug("deleting configurations for ldap server") | |
| cmd = deleteLdapConfiguration.deleteLdapConfigurationCmd() | |
| cmd.hostname = cls.manualLdap.hostname | |
| cmd.port = cls.manualLdap.port | |
| cmd.domainid = cls.manualLdap.domainid | |
| response = cls.apiclient.deleteLdapConfiguration(cmd) | |
| cls.logger.debug("configuration deleted for %s" % response) | |
| cmd.hostname = cls.importLdap.hostname | |
| cmd.port = cls.importLdap.port | |
| cmd.domainid = cls.importLdap.domainid | |
| response = cls.apiclient.deleteLdapConfiguration(cmd) | |
| cls.logger.debug("configuration deleted for %s" % response) | |
| cmd.hostname = cls.syncLdap.hostname | |
| cmd.port = cls.syncLdap.port | |
| cmd.domainid = cls.syncLdap.domainid | |
| cls.logger.debug("deleting configuration %s" % cmd) | |
| response = cls.apiclient.deleteLdapConfiguration(cmd) | |
| cls.logger.debug("configuration deleted for %s" % response) | |
| @classmethod | |
| def create_sync_accounts(cls): | |
| cls.logger.debug("creating account: %s" % LdapTestData.seniors) | |
| cls.senior_account = cls.bind_account_to_ldap( | |
| account=cls.testdata.testdata[LdapTestData.syncAccounts][1]["name"], | |
| ldapdomain=cls.testdata.testdata[LdapTestData.syncAccounts][1]["group"], | |
| accounttype=cls.testdata.testdata[LdapTestData.syncAccounts][1]["accounttype"]) | |
| cls.junior_account = cls.bind_account_to_ldap( | |
| account=cls.testdata.testdata[LdapTestData.syncAccounts][0]["name"], | |
| ldapdomain=cls.testdata.testdata[LdapTestData.syncAccounts][0]["group"], | |
| accounttype=cls.testdata.testdata[LdapTestData.syncAccounts][0]["accounttype"]) | |
| @classmethod | |
| def bind_account_to_ldap(cls, account, ldapdomain, type="Group", accounttype=0): | |
| cmd = linkAccountToLdap.linkAccountToLdapCmd() | |
| cmd.domainid = cls.syncDomain.id | |
| cmd.account = account | |
| cmd.ldapdomain = ldapdomain | |
| cmd.type = type | |
| cmd.accounttype = accounttype | |
| response = cls.apiclient.linkAccountToLdap(cmd) | |
| cls.logger.info("account linked to ladp %s" % response) | |
| # this is needed purely for cleanup: | |
| response = Account.list(cls.apiclient, id=response.accountid) | |
| account_created = Account(response[0].__dict__) | |
| cls._cleanup.append(account_created) | |
| return account_created | |
| @classmethod | |
| def bind_domain_to_ldap(cls, domainid, ldapdomain, type="OU", accounttype=0): | |
| cmd = linkDomainToLdap.linkDomainToLdapCmd() | |
| cmd.domainid = domainid | |
| cmd.type = type | |
| cmd.accounttype = accounttype | |
| cmd.ldapdomain = ldapdomain | |
| response = cls.apiclient.linkDomainToLdap(cmd) | |
| cls.logger.info("domain linked to ladp %s" % response) | |
| @classmethod | |
| def set_ldap_settings_on_domain(cls, domainid): | |
| cmd = updateConfiguration.updateConfigurationCmd() | |
| cmd.domainid = domainid | |
| cmd.name = LdapTestData.basednConfig | |
| cmd.value = cls.testdata.testdata[LdapTestData.configuration][LdapTestData.basedn] | |
| response = cls.apiclient.updateConfiguration(cmd) | |
| cls.logger.debug("set the basedn: %s" % response) | |
| cmd.name = LdapTestData.ldapPwConfig | |
| cmd.value = cls.testdata.testdata[LdapTestData.configuration][LdapTestData.ldapPw] | |
| response = cls.apiclient.updateConfiguration(cmd) | |
| cls.logger.debug("set the pw: %s" % response) | |
| cmd.name = LdapTestData.principalConfig | |
| cmd.value = cls.testdata.testdata[LdapTestData.configuration][LdapTestData.principal] | |
| response = cls.apiclient.updateConfiguration(cmd) | |
| cls.logger.debug("set the id: %s" % response) | |
| if LdapTestData.groupPrinciple in cls.testdata.testdata[LdapTestData.configuration] : | |
| cmd.name = LdapTestData.groupPrinciple | |
| cmd.value = cls.testdata.testdata[LdapTestData.configuration][LdapTestData.groupPrinciple] | |
| response = cls.apiclient.updateConfiguration(cmd) | |
| cls.logger.debug("set the id: %s" % response) | |
| ## python ldap utility functions |