| # 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) |
