| # 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): |
| try: |
| cleanup_resources(cls.api_client, cls._cleanup) |
| |
| except Exception as tde: |
| raise Exception("Warning: Exception during cleanup : %s" % tde) |
| return |
| |
| 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 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 |
