| # 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. | |
| """ P1 for LDAP Config | |
| """ | |
| #!/usr/bin/env python | |
| from marvin.cloudstackTestCase import cloudstackTestCase | |
| from marvin.cloudstackAPI import (updateConfiguration, | |
| createAccount, | |
| deleteAccount, | |
| addLdapConfiguration, | |
| deleteLdapConfiguration) | |
| from marvin.cloudstackAPI import login | |
| from marvin.lib.utils import cleanup_resources | |
| from nose.plugins.attrib import attr | |
| import telnetlib | |
| from ddt import ddt, data | |
| @ddt | |
| class TestLdap(cloudstackTestCase): | |
| """ | |
| This tests attempts to register a LDAP server and authenticate as an LDAP user. | |
| """ | |
| @classmethod | |
| def setUpClass(cls): | |
| testClient = super(TestLdap, cls).getClsTestClient() | |
| cls.api_client = testClient.getApiClient() | |
| cls.services = testClient.getParsedTestDataConfig() | |
| cls._cleanup = [] | |
| cls.delflag = 0 | |
| cls.reason = "" | |
| @classmethod | |
| def tearDownClass(cls): | |
| super(TestLdap, cls).tearDownClass() | |
| def setUp(self): | |
| self.apiClient = self.testClient.getApiClient() | |
| self.acct = createAccount.createAccountCmd() | |
| self.acct.accounttype = 0 | |
| self.acct.firstname = self.services[ | |
| "configurableData"]["ldap_account"]["firstname"] | |
| self.acct.lastname = self.services[ | |
| "configurableData"]["ldap_account"]["lastname"] | |
| self.acct.password = self.services[ | |
| "configurableData"]["ldap_account"]["password"] | |
| self.acct.username = self.services[ | |
| "configurableData"]["ldap_account"]["username"] | |
| self.acct.email = self.services[ | |
| "configurableData"]["ldap_account"]["email"] | |
| self.acct.account = self.services[ | |
| "configurableData"]["ldap_account"]["username"] | |
| self.acct.domainid = 1 | |
| if self.acct.firstname == ""or self.acct.lastname == "" or self.acct.password == "" or self.acct.username == ""\ | |
| or self.services["configurableData"]["ldap_configuration"]["ldapUsername"] == "" or \ | |
| self.acct.account == "" or \ | |
| self.services["configurableData"]["ldap_configuration"]["ldapPassword"] == "": | |
| self.debug("Please rerun the test by providing values in ldap configuration user details") | |
| self.skipTest("Please rerun the test by providing proper values in configuration file") | |
| else: | |
| self.delflag = 1 | |
| self.acctRes = self.apiClient.createAccount(self.acct) | |
| self.assertEqual(self.delflag, 1, "LDAP account details are not provided,please check the configuration") | |
| return | |
| def tearDown(self): | |
| self.debug("In tear down%s" % self.delflag) | |
| try: | |
| deleteAcct = deleteAccount.deleteAccountCmd() | |
| deleteAcct.id = self.acctRes.id | |
| acct_name = self.acctRes.name | |
| self.apiClient.deleteAccount(deleteAcct) | |
| self.debug( | |
| "Deleted the following account name %s:" % | |
| acct_name) | |
| if self.ldapconfRes == 1: | |
| self._deleteldapconfiguration( | |
| self.services["configurableData"]["ldap_configuration"]) | |
| except Exception as e: | |
| raise Exception("Warning: Exception during cleanup : %s" % e) | |
| return | |
| @attr(tags=["advanced", "basic"], required_hardware="false") | |
| def test_01_addLdapConfiguration(self): | |
| """ | |
| This test configures LDAP and attempts to authenticate as a user. | |
| """ | |
| self.ldapconfRes = self._addLdapConfiguration( | |
| self.services["configurableData"]["ldap_configuration"]) | |
| if self.ldapconfRes == 1: | |
| self.debug("Ldap Configuration was successful") | |
| loginRes = self._checklogin( | |
| self.services["configurableData"]["ldap_configuration"]["ldapUsername"], | |
| self.services["configurableData"]["ldap_configuration"]["ldapPassword"]) | |
| self.debug(loginRes) | |
| self.assertEqual(loginRes, 1, self.reason) | |
| else: | |
| self.debug("LDAP Configuration failed with exception") | |
| self.assertEqual( | |
| self.ldapconfRes, | |
| 1, | |
| self.reason) | |
| def test_02_validateldapsecuritypatch(self): | |
| self.ldapconfRes = self._addLdapConfiguration( | |
| self.services["configurableData"]["ldap_configuration"]) | |
| self.assertEqual(self.ldapconfRes, 1, "Ldap Configuration failed") | |
| loginRes = self._checklogin( | |
| self.services["configurableData"]["ldap_configuration"]["ldapUsername"], "") | |
| self.assertNotEqual(loginRes, 1, "login API Successful with empty password") | |
| @data("basedn", "ldapPassword") | |
| def test_03_validateldapbindnobasedn(self, value): | |
| """ | |
| This test is to verify ldapbind functionality without passing required bind parameters. | |
| """ | |
| bindvalue = self.services["configurableData"]["ldap_configuration"][value] | |
| if len(bindvalue) > 0: | |
| self.services["configurableData"]["ldap_configuration"][value] = "" | |
| self.ldapconfRes = self._addLdapConfiguration(self.services["configurableData"]["ldap_configuration"]) | |
| if self.reason.__contains__("addLdapConfiguration failed"): | |
| self.assertEqual(self.ldapconfRes, 1, "Ldap Configuration not successful") | |
| else: | |
| self.assertNotEqual(self.ldapconfRes, 1, "Ldap Configuration successful with invalid values-i.e." | |
| " allowing anonymous bind") | |
| self.services["configurableData"]["ldap_configuration"][value] = value | |
| def _addLdapConfiguration(self, ldapConfiguration): | |
| """ | |
| :param ldapConfiguration | |
| """ | |
| self.chkConfig = self._checkLdapConfiguration(ldapConfiguration) | |
| if not self.chkConfig: | |
| return 0 | |
| # Setup Global settings | |
| updateConfigurationCmd = updateConfiguration.updateConfigurationCmd() | |
| updateConfigurationCmd.name = "ldap.basedn" | |
| updateConfigurationCmd.value = ldapConfiguration['basedn'] | |
| updateConfigurationResponse = self.apiClient.updateConfiguration( | |
| updateConfigurationCmd) | |
| self.debug( | |
| "updated the parameter %s with value %s" % | |
| (updateConfigurationResponse.name, | |
| updateConfigurationResponse.value)) | |
| updateConfigurationCmd = updateConfiguration.updateConfigurationCmd() | |
| updateConfigurationCmd.name = "ldap.email.attribute" | |
| updateConfigurationCmd.value = ldapConfiguration['emailAttribute'] | |
| updateConfigurationResponse = self.apiClient.updateConfiguration( | |
| updateConfigurationCmd) | |
| self.debug( | |
| "updated the parameter %s with value %s" % | |
| (updateConfigurationResponse.name, | |
| updateConfigurationResponse.value)) | |
| updateConfigurationCmd = updateConfiguration.updateConfigurationCmd() | |
| updateConfigurationCmd.name = "ldap.user.object" | |
| updateConfigurationCmd.value = ldapConfiguration['userObject'] | |
| updateConfigurationResponse = self.apiClient.updateConfiguration( | |
| updateConfigurationCmd) | |
| self.debug( | |
| "updated the parameter %s with value %s" % | |
| (updateConfigurationResponse.name, | |
| updateConfigurationResponse.value)) | |
| updateConfigurationCmd = updateConfiguration.updateConfigurationCmd() | |
| updateConfigurationCmd.name = "ldap.username.attribute" | |
| updateConfigurationCmd.value = ldapConfiguration['usernameAttribute'] | |
| updateConfigurationResponse = self.apiClient.updateConfiguration( | |
| updateConfigurationCmd) | |
| self.debug( | |
| "updated the parameter %s with value %s" % | |
| (updateConfigurationResponse.name, | |
| updateConfigurationResponse.value)) | |
| self.debug("start addLdapConfiguration test") | |
| ldapServer = addLdapConfiguration.addLdapConfigurationCmd() | |
| ldapServer.hostname = ldapConfiguration['hostname'] | |
| ldapServer.port = ldapConfiguration['port'] | |
| self.debug("calling addLdapConfiguration API command") | |
| try: | |
| self.apiClient.addLdapConfiguration(ldapServer) | |
| self.debug("addLdapConfiguration was successful") | |
| return 1 | |
| except Exception as e: | |
| self.debug("addLdapConfiguration failed %s Check the Passed passed ldap attributes" % e) | |
| self.reason = "addLdapConfiguration failed %s Check the Passed passed ldap attributes" % e | |
| return 0 | |
| def _checkLdapConfiguration(self, ldapConfiguration): | |
| """" | |
| This function checks 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")) | |
| self.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: | |
| self.debug(" Not able to reach the LDAP server ,please check the Services on LDAP %s and exception is %s" | |
| % ((ldapConfiguration['hostname']), e)) | |
| self.reason = "Not able to reach the LDAP server ,please check the Services on LDAP %s and exception is %s"\ | |
| % ((ldapConfiguration['hostname']), e) | |
| return flag | |
| def _deleteldapconfiguration(self, ldapConfiguration): | |
| """ | |
| :param ldapConfiguration | |
| """ | |
| ldapServer = deleteLdapConfiguration.deleteLdapConfigurationCmd() | |
| ldapServer.hostname = ldapConfiguration["hostname"] | |
| try: | |
| self.apiClient.deleteLdapConfiguration(ldapServer) | |
| self.debug("deleteLdapConfiguration was successful") | |
| return 1 | |
| except Exception as e: | |
| self.debug("deleteLdapConfiguration failed %s" % e) | |
| return 0 | |
| def _checklogin(self, username, password): | |
| """ | |
| :param username: | |
| :param password: | |
| """ | |
| self.debug("Attempting to login.") | |
| try: | |
| loginParams = login.loginCmd() | |
| loginParams.username = username | |
| loginParams.password = password | |
| loginRes = self.apiClient.login(loginParams) | |
| self.debug("login response %s" % loginRes) | |
| if loginRes is None: | |
| self.debug("login not successful") | |
| return 0 | |
| else: | |
| self.debug("login successful") | |
| return 1 | |
| except Exception as p: | |
| self.debug("login operation failed %s" % p) | |
| self.reason = "Login operation Failed %s" % p |