# 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.cloudstackTestCase import cloudstackTestCase from marvin.cloudstackAPI import ( updateConfiguration, deleteAccount, addLdapConfiguration, linkDomainToLdap, deleteLdapConfiguration, disableAccount) from marvin.lib.common import get_domain from marvin.lib.base import (Account, Configurations, Domain) from marvin.cloudstackAPI import login from marvin.lib.utils import (cleanup_resources) from nose.plugins.attrib import attr import telnetlib import random import string def randomword(length): return ''.join(random.choice(string.lowercase) for i in range(length)) def addLdapConfiguration1(cls, ldapConfiguration): """ :param ldapConfiguration """ cls.chkConfig = checkLdapConfiguration(cls, ldapConfiguration) if not cls.chkConfig: return 0 # Setup Global settings Configurations.update( cls.apiClient, name="ldap.basedn", value=ldapConfiguration['basedn'] ) Configurations.update( cls.apiClient, name="ldap.bind.password", value=ldapConfiguration['bindpassword'] ) Configurations.update( cls.apiClient, name="ldap.bind.principal", value=ldapConfiguration['principal'] ) Configurations.update( cls.apiClient, name="ldap.email.attribute", value=ldapConfiguration['emailAttribute'] ) Configurations.update( cls.apiClient, name="ldap.user.object", value=ldapConfiguration['userObject'] ) Configurations.update( cls.apiClient, name="ldap.username.attribute", value=ldapConfiguration['usernameAttribute'] ) Configurations.update( cls.apiClient, name="ldap.nested.groups.enable", value="true" ) ldapServer = addLdapConfiguration.addLdapConfigurationCmd() ldapServer.hostname = ldapConfiguration['hostname'] ldapServer.port = ldapConfiguration['port'] cls.debug("calling addLdapConfiguration API command") try: cls.apiClient.addLdapConfiguration(ldapServer) cls.debug("addLdapConfiguration was successful") return 1 except Exception as e: cls.debug( "addLdapConfiguration failed %s Check the Passed passed" " ldap attributes" % e) cls.reason = "addLdapConfiguration failed %s Check the Passed " \ "passed ldap attributes" % e raise Exception( "addLdapConfiguration failed %s Check the Passed passed" " ldap attributes" % e) return 1 def checklogin(cls, username, password, domain, method): """ :param username: :param password: """ cls.debug("Attempting to login.") try: loginParams = login.loginCmd() loginParams.username = username loginParams.password = password loginParams.domain = domain loginRes = cls.apiClient.login(loginParams, method) cls.debug("login response %s" % loginRes) if loginRes is None: cls.debug("login not successful") return 0 else: cls.debug("login successful") return 1 except Exception as p: cls.debug("login operation failed %s" % p) cls.reason = "Login operation Failed %s" % p def checkLdapConfiguration(cls, ldapConfiguration): """This function checks whether the passed ldap server in the configuration is up and running or not. """ flag = False try: tn = telnetlib.Telnet( ldapConfiguration['hostname'], ldapConfiguration['port'], timeout=15) if tn is not None: tn.set_debuglevel(1) print(tn.msg("Connected to the server")) cls.debug( "Ldap Server is Up and listening on the port %s" % tn.msg("Connected to the server")) flag = True tn.close() except Exception as e: cls.debug( "Not able to reach the LDAP server ," "please check the Services on LDAP %s and exception is %s" % ((ldapConfiguration['hostname']), e)) cls.reason = "Not able to reach the LDAP server ,please check" \ " the Services on LDAP %s and exception is %s" \ % ((ldapConfiguration['hostname']), e) return flag class TestLdap(cloudstackTestCase): """ LDAP AutoImport smoke tests """ @classmethod def setUpClass(cls): """ :type cls: object """ testClient = super(TestLdap, cls).getClsTestClient() cls.api_client = testClient.getApiClient() cls.services = testClient.getParsedTestDataConfig() cls.cleanup = [] cls.domain = get_domain(cls.api_client) cls.delflag = 0 cls.reason = "" cls.apiClient = cls.testClient.getApiClient() try: cls.ldapconfRes = addLdapConfiguration1( cls, cls.services["configurableData"]["ldap_configuration"]) except Exception as e: raise Exception("Configuring LDAP failed. Check attributes") cls.cleanup.append(cls.ldapconfRes) @classmethod def tearDownClass(cls): """ #cleanup includes : delete normal account, remove ldap configuration :type cls: object """ testClient = super(TestLdap, cls).getClsTestClient() cls.api_client = testClient.getApiClient() cls.services = testClient.getParsedTestDataConfig() if cls.ldapconfRes == 1: ldapserver = deleteLdapConfiguration.deleteLdapConfigurationCmd() ldapserver.hostname = cls.services["configurableData"][ "ldap_configuration"]["hostname"] try: cls.apiClient.deleteLdapConfiguration(ldapserver) cls.debug("deleteLdapConfiguration was successful") return 1 except Exception as e: cls.debug("deleteLdapConfiguration failed %s" % e) return 0 def setUp(self): self.user = self.services["configurableData"]["link_ldap_details"]["linkLdapUsername"] self.password = self.services["configurableData"]["link_ldap_details"]["linkLdapPassword"] self.delflag1 = 0 self.delflag2 = 0 self.delflag3 = 0 self.delflag4 = 0 self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.cleanup = [] self.parent_domain = Domain.create( self.apiclient, services=self.services["domain"], parentdomainid=self.domain.id) self.ldaplink = linkDomainToLdap.linkDomainToLdapCmd() self.ldaplink.domainid = self.parent_domain.id self.ldaplink.accounttype = self.services[ "configurableData"]["link_ldap_details"]["accounttype"] self.ldaplink.name = self.services[ "configurableData"]["link_ldap_details"]["name"] self.ldaplink.type = self.services[ "configurableData"]["link_ldap_details"]["type"] if self.services["configurableData"][ "link_ldap_details"]["admin"] is not None: self.ldaplink.admin = self.services[ "configurableData"]["link_ldap_details"]["admin"] if self.ldaplink.domainid == "" or self.ldaplink.accounttype == "" \ or self.ldaplink.name == "" \ or self.ldaplink.type == "": self.debug( "Please rerun the test by providing " "values in link_ldap configuration user details") self.skipTest( "Please rerun the test by providing " "proper values in configuration file(link ldap)") else: self.delflag1 = 1 self.ldaplinkRes = self.apiClient.linkDomainToLdap(self.ldaplink) self.assertEqual( self.delflag1, 1, "Linking LDAP failed,please check the configuration") loginRes = checklogin(self, self.user, self.password, self.parent_domain.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, 1, self.reason) lsap_user = Account.list(self.api_client, domainid=self.parent_domain.id, name=self.user ) self.ldapacctID = lsap_user[0].id def tearDown(self): try: self.parent_domain.delete(self.apiclient, cleanup=True) except Exception as e: raise Exception( "Warning: Exception during cleanup of domain : %s" % e) try: # Clean up, terminate the created instance, volumes and snapshots cleanup_resources(self.apiclient, self.cleanup) pass except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return @attr(tags=["advanced", "basic"], required_hardware="true") def test_01_ldap(self): """Check the linkDomainToLdap functionality""" self.domain1 = Domain.create( self.apiclient, services=self.services["domain"], parentdomainid=self.domain.id) self.ldaplink4 = linkDomainToLdap.linkDomainToLdapCmd() self.ldaplink4.domainid = self.domain1.id self.ldaplink4.accounttype = self.services[ "configurableData"]["link_ldap_details"]["accounttype"] self.ldaplink4.name = self.services[ "configurableData"]["link_ldap_details"]["name"] self.ldaplink4.type = self.services[ "configurableData"]["link_ldap_details"]["type"] if self.services["configurableData"][ "link_ldap_details"]["admin"] is not None: self.ldaplink4.admin = self.services[ "configurableData"]["link_ldap_details"]["admin"] try: self.ldaplinkRes4 = self.apiClient.linkDomainToLdap(self.ldaplink4) except Exception as e: raise Exception( "Linking LDAP failed,please check the configuration") try: self.domain1.delete(self.apiclient) except Exception as e: raise Exception( "Warning: Exception during deletion of domain : %s" % e) @attr(tags=["advanced", "basic"], required_hardware="true") def test_02_ldap(self): """User is both in LDAP and imported into CS(i.e already logged in once.So just check the log in again)""" loginRes = checklogin( self, self.user, self.password, self.parent_domain.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, 1, self.reason) @attr(tags=["advanced", "basic"], required_hardware="true") def test_03_ldap(self): """User in LDAP, wrong password --> login should fail""" loginRes = checklogin( self, self.user, randomword(8), self.parent_domain.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, None, self.reason) @attr(tags=["advanced", "basic"], required_hardware="true") def test_04_ldap(self): """User is only present locally, password is wrong --> login should fail""" loginRes = checklogin( self, self.services["configurableData"]["ldap_account"]["username"], randomword(10), "", method="POST") self.debug(loginRes) self.assertEqual(loginRes, None, self.reason) @attr(tags=["advanced", "basic"], required_hardware="true") def test_05_ldap(self): """user is not present anywhere --> login should fail""" loginRes = checklogin(self, randomword(10), randomword(10), self.parent_domain.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, None, self.reason) @attr(tags=["advanced", "basic"], required_hardware="true") def test_06_ldap(self): """Delete the LDAP user from CS and try to login --> User should be created again""" try: deleteAcct2 = deleteAccount.deleteAccountCmd() deleteAcct2.id = self.ldapacctID acct_name = self.services["configurableData"][ "link_ldap_details"]["linkLdapUsername"] self.apiClient.deleteAccount(deleteAcct2) self.debug( "Deleted the following account name %s:" % acct_name) except Exception as e: raise Exception( "Warning: Exception during deleting " "ldap imported account : %s" % e) loginRes = checklogin( self, self.user, self.password, self.parent_domain.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, 1, self.reason) @attr(tags=["advanced", "basic"], required_hardware="true") def test_07_ldap(self): """Lock the user from CS and attempt to login --> login should fail""" self.lockAcct = disableAccount.disableAccountCmd() self.lockAcct.lock = 'true' self.lockAcct.account = self.services["configurableData"][ "ldap_account"]["username"] self.lockAcct.domainid = self.parent_domain.id self.apiClient.disableAccount(self.lockAcct) loginRes = checklogin( self, self.user, self.password, self.parent_domain.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, None, self.reason) @attr(tags=["advanced", "basic"], required_hardware="true") def test_08_ldap(self): """Create different domains and link all of them to LDAP. Check login in each domain --> login should be successful""" try: loginRes = checklogin( self, self.user, self.password, self.parent_domain.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, 1, self.reason) self.domain2 = Domain.create( self.apiclient, services=self.services["domain"], parentdomainid=self.domain.id) # here link ldap to domain self.ldaplink2 = linkDomainToLdap.linkDomainToLdapCmd() self.ldaplink2.domainid = self.domain2.id self.ldaplink2.accounttype = self.services[ "configurableData"]["link_ldap_details"]["accounttype"] self.ldaplink2.name = self.services[ "configurableData"]["link_ldap_details"]["name"] self.ldaplink2.type = self.services[ "configurableData"]["link_ldap_details"]["type"] if self.services["configurableData"][ "link_ldap_details"]["admin"] is not None: self.ldaplink2.admin = self.services[ "configurableData"]["link_ldap_details"]["admin"] if self.ldaplink2.domainid == "" \ or self.ldaplink2.accounttype == "" \ or self.ldaplink2.name == "" \ or self.ldaplink2.type == "": self.debug( "Please rerun the test by providing" " values in link_ldap configuration user details") self.skipTest( "Please rerun the test by providing " "proper values in configuration file(link ldap)") else: self.delflag2 = 1 self.ldaplinkRes2 = self.apiClient.linkDomainToLdap( self.ldaplink2) self.assertEqual( self.delflag2, 1, "Linking LDAP failed,please check the configuration") loginRes = checklogin( self, self.user, self.password, self.domain2.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, 1, self.reason) self.domain3 = Domain.create( self.apiclient, services=self.services["domain"], parentdomainid=self.domain.id) # here link ldap to domain self.ldaplink3 = linkDomainToLdap.linkDomainToLdapCmd() self.ldaplink3.domainid = self.domain3.id self.ldaplink3.accounttype = self.services[ "configurableData"]["link_ldap_details"]["accounttype"] self.ldaplink3.name = self.services[ "configurableData"]["link_ldap_details"]["name"] self.ldaplink3.type = self.services[ "configurableData"]["link_ldap_details"]["type"] if self.services["configurableData"][ "link_ldap_details"]["admin"] is not None: self.ldaplink3.admin = self.services[ "configurableData"]["link_ldap_details"]["admin"] if self.ldaplink3.domainid == "" \ or self.ldaplink3.accounttype == "" \ or self.ldaplink3.name == "" \ or self.ldaplink3.type == "": self.debug( "Please rerun the test by providing" " values in link_ldap configuration user details") self.skipTest( "Please rerun the test by providing " "proper values in configuration file(link ldap)") else: self.delflag3 = 1 self.ldaplinkRes3 = self.apiClient.linkDomainToLdap( self.ldaplink3) self.assertEqual( self.delflag3, 1, "Linking LDAP failed,please check the configuration") loginRes = checklogin( self, self.user, self.password, self.domain2.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, 1, self.reason) finally: try: self.domain2.delete(self.apiclient, cleanup=True) except Exception as e: raise Exception( "Warning: Exception during deletion of domain : %s" % e) try: self.domain3.delete(self.apiclient, cleanup=True) except Exception as e: raise Exception( "Warning: Exception during deletion of domain : %s" % e) return @attr(tags=["advanced", "basic"], required_hardware="true") def test_09_ldap(self): """ Enable nested groups and try to login with a user that is in nested group --> login should be successful""" if self.services["configurableData"]["link_ldap_details"]["linkLdapNestedUser"] == "": self.skipTest("No nested user mentioned") updateConfigurationCmd = updateConfiguration.updateConfigurationCmd() updateConfigurationCmd.name = "ldap.nested.groups.enable" updateConfigurationCmd.value = 'true' self.apiClient.updateConfiguration(updateConfigurationCmd) loginRes = checklogin( self, self.services["configurableData"]["link_ldap_details"]["linkLdapNestedUser"], self.services["configurableData"]["link_ldap_details"]["linkLdapNestedPassword"], self.parent_domain.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, 1, self.reason) @attr(tags=["advanced", "basic"], required_hardware="true") def test_10_ldap(self): """Check db tables""" db_check = 1 domainID = self.dbclient.execute( "SELECT id FROM domain WHERE uuid=" + "'" + self.parent_domain.id + "'" + ";", db="cloud") dbChecking = self.dbclient.execute( "SELECT type,name,account_type " "FROM ldap_trust_map WHERE domain_id=" + "'" + str(domainID[0][0]) + "'" + ";", db="cloud") if dbChecking is not None and str( dbChecking[0][0]) == \ self.services["configurableData"]["link_ldap_details"]["type"] \ and str( dbChecking[0][1]) == \ self.services["configurableData"]["link_ldap_details"]["name"] \ and str( dbChecking[0][2]) == \ self.services["configurableData"]["link_ldap_details"]["accounttype"]: db_check = 0 self.assertEqual(db_check, 0, "DB check failed") @attr(tags=["advanced", "basic"], required_hardware="true") def test_11_ldap(self): """Password/domain empty --> login should fail""" loginRes = checklogin( self, "", "", self.parent_domain.name, method="POST") self.debug(loginRes) self.assertEqual(loginRes, None, self.reason)