aboutsummaryrefslogtreecommitdiffstats
path: root/keystone-moon/keystone/tests/unit/test_backend_ldap.py
diff options
context:
space:
mode:
Diffstat (limited to 'keystone-moon/keystone/tests/unit/test_backend_ldap.py')
-rw-r--r--keystone-moon/keystone/tests/unit/test_backend_ldap.py3049
1 files changed, 3049 insertions, 0 deletions
diff --git a/keystone-moon/keystone/tests/unit/test_backend_ldap.py b/keystone-moon/keystone/tests/unit/test_backend_ldap.py
new file mode 100644
index 00000000..10119808
--- /dev/null
+++ b/keystone-moon/keystone/tests/unit/test_backend_ldap.py
@@ -0,0 +1,3049 @@
+# -*- coding: utf-8 -*-
+# Copyright 2012 OpenStack Foundation
+# Copyright 2013 IBM Corp.
+#
+# Licensed 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 copy
+import uuid
+
+import ldap
+import mock
+from oslo_config import cfg
+from testtools import matchers
+
+from keystone.common import cache
+from keystone.common import ldap as common_ldap
+from keystone.common.ldap import core as common_ldap_core
+from keystone.common import sql
+from keystone import exception
+from keystone import identity
+from keystone.identity.mapping_backends import mapping as map
+from keystone import resource
+from keystone.tests import unit as tests
+from keystone.tests.unit import default_fixtures
+from keystone.tests.unit import fakeldap
+from keystone.tests.unit import identity_mapping as mapping_sql
+from keystone.tests.unit.ksfixtures import database
+from keystone.tests.unit import test_backend
+
+
+CONF = cfg.CONF
+
+
+def create_group_container(identity_api):
+ # Create the groups base entry (ou=Groups,cn=example,cn=com)
+ group_api = identity_api.driver.group
+ conn = group_api.get_connection()
+ dn = 'ou=Groups,cn=example,cn=com'
+ conn.add_s(dn, [('objectclass', ['organizationalUnit']),
+ ('ou', ['Groups'])])
+
+
+class BaseLDAPIdentity(test_backend.IdentityTests):
+
+ def setUp(self):
+ super(BaseLDAPIdentity, self).setUp()
+ self.clear_database()
+
+ common_ldap.register_handler('fake://', fakeldap.FakeLdap)
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+
+ self.addCleanup(common_ldap_core._HANDLERS.clear)
+
+ def _get_domain_fixture(self):
+ """Domains in LDAP are read-only, so just return the static one."""
+ return self.resource_api.get_domain(CONF.identity.default_domain_id)
+
+ def clear_database(self):
+ for shelf in fakeldap.FakeShelves:
+ fakeldap.FakeShelves[shelf].clear()
+
+ def reload_backends(self, domain_id):
+ # Only one backend unless we are using separate domain backends
+ self.load_backends()
+
+ def get_config(self, domain_id):
+ # Only one conf structure unless we are using separate domain backends
+ return CONF
+
+ def config_overrides(self):
+ super(BaseLDAPIdentity, self).config_overrides()
+ self.config_fixture.config(
+ group='identity',
+ driver='keystone.identity.backends.ldap.Identity')
+
+ def config_files(self):
+ config_files = super(BaseLDAPIdentity, self).config_files()
+ config_files.append(tests.dirs.tests_conf('backend_ldap.conf'))
+ return config_files
+
+ def get_user_enabled_vals(self, user):
+ user_dn = (
+ self.identity_api.driver.user._id_to_dn_string(user['id']))
+ enabled_attr_name = CONF.ldap.user_enabled_attribute
+
+ ldap_ = self.identity_api.driver.user.get_connection()
+ res = ldap_.search_s(user_dn,
+ ldap.SCOPE_BASE,
+ u'(sn=%s)' % user['name'])
+ if enabled_attr_name in res[0][1]:
+ return res[0][1][enabled_attr_name]
+ else:
+ return None
+
+ def test_build_tree(self):
+ """Regression test for building the tree names
+ """
+ user_api = identity.backends.ldap.UserApi(CONF)
+ self.assertTrue(user_api)
+ self.assertEqual("ou=Users,%s" % CONF.ldap.suffix, user_api.tree_dn)
+
+ def test_configurable_allowed_user_actions(self):
+ user = {'name': u'fäké1',
+ 'password': u'fäképass1',
+ 'domain_id': CONF.identity.default_domain_id,
+ 'tenants': ['bar']}
+ user = self.identity_api.create_user(user)
+ self.identity_api.get_user(user['id'])
+
+ user['password'] = u'fäképass2'
+ self.identity_api.update_user(user['id'], user)
+
+ self.identity_api.delete_user(user['id'])
+ self.assertRaises(exception.UserNotFound,
+ self.identity_api.get_user,
+ user['id'])
+
+ def test_configurable_forbidden_user_actions(self):
+ conf = self.get_config(CONF.identity.default_domain_id)
+ conf.ldap.user_allow_create = False
+ conf.ldap.user_allow_update = False
+ conf.ldap.user_allow_delete = False
+ self.reload_backends(CONF.identity.default_domain_id)
+
+ user = {'name': u'fäké1',
+ 'password': u'fäképass1',
+ 'domain_id': CONF.identity.default_domain_id,
+ 'tenants': ['bar']}
+ self.assertRaises(exception.ForbiddenAction,
+ self.identity_api.create_user,
+ user)
+
+ self.user_foo['password'] = u'fäképass2'
+ self.assertRaises(exception.ForbiddenAction,
+ self.identity_api.update_user,
+ self.user_foo['id'],
+ self.user_foo)
+
+ self.assertRaises(exception.ForbiddenAction,
+ self.identity_api.delete_user,
+ self.user_foo['id'])
+
+ def test_configurable_forbidden_create_existing_user(self):
+ conf = self.get_config(CONF.identity.default_domain_id)
+ conf.ldap.user_allow_create = False
+ self.reload_backends(CONF.identity.default_domain_id)
+
+ self.assertRaises(exception.ForbiddenAction,
+ self.identity_api.create_user,
+ self.user_foo)
+
+ def test_user_filter(self):
+ user_ref = self.identity_api.get_user(self.user_foo['id'])
+ self.user_foo.pop('password')
+ self.assertDictEqual(user_ref, self.user_foo)
+
+ conf = self.get_config(user_ref['domain_id'])
+ conf.ldap.user_filter = '(CN=DOES_NOT_MATCH)'
+ self.reload_backends(user_ref['domain_id'])
+ # invalidate the cache if the result is cached.
+ self.identity_api.get_user.invalidate(self.identity_api,
+ self.user_foo['id'])
+ self.assertRaises(exception.UserNotFound,
+ self.identity_api.get_user,
+ self.user_foo['id'])
+
+ def test_remove_role_grant_from_user_and_project(self):
+ self.assignment_api.create_grant(user_id=self.user_foo['id'],
+ project_id=self.tenant_baz['id'],
+ role_id='member')
+ roles_ref = self.assignment_api.list_grants(
+ user_id=self.user_foo['id'],
+ project_id=self.tenant_baz['id'])
+ self.assertDictEqual(roles_ref[0], self.role_member)
+
+ self.assignment_api.delete_grant(user_id=self.user_foo['id'],
+ project_id=self.tenant_baz['id'],
+ role_id='member')
+ roles_ref = self.assignment_api.list_grants(
+ user_id=self.user_foo['id'],
+ project_id=self.tenant_baz['id'])
+ self.assertEqual(0, len(roles_ref))
+ self.assertRaises(exception.RoleAssignmentNotFound,
+ self.assignment_api.delete_grant,
+ user_id=self.user_foo['id'],
+ project_id=self.tenant_baz['id'],
+ role_id='member')
+
+ def test_get_and_remove_role_grant_by_group_and_project(self):
+ new_domain = self._get_domain_fixture()
+ new_group = {'domain_id': new_domain['id'],
+ 'name': uuid.uuid4().hex}
+ new_group = self.identity_api.create_group(new_group)
+ new_user = {'name': 'new_user', 'enabled': True,
+ 'domain_id': new_domain['id']}
+ new_user = self.identity_api.create_user(new_user)
+ self.identity_api.add_user_to_group(new_user['id'],
+ new_group['id'])
+
+ roles_ref = self.assignment_api.list_grants(
+ group_id=new_group['id'],
+ project_id=self.tenant_bar['id'])
+ self.assertEqual([], roles_ref)
+ self.assertEqual(0, len(roles_ref))
+
+ self.assignment_api.create_grant(group_id=new_group['id'],
+ project_id=self.tenant_bar['id'],
+ role_id='member')
+ roles_ref = self.assignment_api.list_grants(
+ group_id=new_group['id'],
+ project_id=self.tenant_bar['id'])
+ self.assertNotEmpty(roles_ref)
+ self.assertDictEqual(roles_ref[0], self.role_member)
+
+ self.assignment_api.delete_grant(group_id=new_group['id'],
+ project_id=self.tenant_bar['id'],
+ role_id='member')
+ roles_ref = self.assignment_api.list_grants(
+ group_id=new_group['id'],
+ project_id=self.tenant_bar['id'])
+ self.assertEqual(0, len(roles_ref))
+ self.assertRaises(exception.RoleAssignmentNotFound,
+ self.assignment_api.delete_grant,
+ group_id=new_group['id'],
+ project_id=self.tenant_bar['id'],
+ role_id='member')
+
+ def test_get_and_remove_role_grant_by_group_and_domain(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_get_role_assignment_by_domain_not_found(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_del_role_assignment_by_domain_not_found(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_get_and_remove_role_grant_by_user_and_domain(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_get_and_remove_correct_role_grant_from_a_mix(self):
+ self.skipTest('Blocked by bug 1101287')
+
+ def test_get_and_remove_role_grant_by_group_and_cross_domain(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_get_and_remove_role_grant_by_user_and_cross_domain(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_role_grant_by_group_and_cross_domain_project(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_role_grant_by_user_and_cross_domain_project(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_multi_role_grant_by_user_group_on_project_domain(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_delete_role_with_user_and_group_grants(self):
+ self.skipTest('Blocked by bug 1101287')
+
+ def test_delete_user_with_group_project_domain_links(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_delete_group_with_user_project_domain_links(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_list_projects_for_user(self):
+ domain = self._get_domain_fixture()
+ user1 = {'name': uuid.uuid4().hex, 'password': uuid.uuid4().hex,
+ 'domain_id': domain['id'], 'enabled': True}
+ user1 = self.identity_api.create_user(user1)
+ user_projects = self.assignment_api.list_projects_for_user(user1['id'])
+ self.assertThat(user_projects, matchers.HasLength(0))
+
+ # new grant(user1, role_member, tenant_bar)
+ self.assignment_api.create_grant(user_id=user1['id'],
+ project_id=self.tenant_bar['id'],
+ role_id=self.role_member['id'])
+ # new grant(user1, role_member, tenant_baz)
+ self.assignment_api.create_grant(user_id=user1['id'],
+ project_id=self.tenant_baz['id'],
+ role_id=self.role_member['id'])
+ user_projects = self.assignment_api.list_projects_for_user(user1['id'])
+ self.assertThat(user_projects, matchers.HasLength(2))
+
+ # Now, check number of projects through groups
+ user2 = {'name': uuid.uuid4().hex, 'password': uuid.uuid4().hex,
+ 'domain_id': domain['id'], 'enabled': True}
+ user2 = self.identity_api.create_user(user2)
+
+ group1 = {'name': uuid.uuid4().hex, 'domain_id': domain['id']}
+ group1 = self.identity_api.create_group(group1)
+
+ self.identity_api.add_user_to_group(user2['id'], group1['id'])
+
+ # new grant(group1(user2), role_member, tenant_bar)
+ self.assignment_api.create_grant(group_id=group1['id'],
+ project_id=self.tenant_bar['id'],
+ role_id=self.role_member['id'])
+ # new grant(group1(user2), role_member, tenant_baz)
+ self.assignment_api.create_grant(group_id=group1['id'],
+ project_id=self.tenant_baz['id'],
+ role_id=self.role_member['id'])
+ user_projects = self.assignment_api.list_projects_for_user(user2['id'])
+ self.assertThat(user_projects, matchers.HasLength(2))
+
+ # new grant(group1(user2), role_other, tenant_bar)
+ self.assignment_api.create_grant(group_id=group1['id'],
+ project_id=self.tenant_bar['id'],
+ role_id=self.role_other['id'])
+ user_projects = self.assignment_api.list_projects_for_user(user2['id'])
+ self.assertThat(user_projects, matchers.HasLength(2))
+
+ def test_list_projects_for_user_and_groups(self):
+ domain = self._get_domain_fixture()
+ # Create user1
+ user1 = {'name': uuid.uuid4().hex, 'password': uuid.uuid4().hex,
+ 'domain_id': domain['id'], 'enabled': True}
+ user1 = self.identity_api.create_user(user1)
+
+ # Create new group for user1
+ group1 = {'name': uuid.uuid4().hex, 'domain_id': domain['id']}
+ group1 = self.identity_api.create_group(group1)
+
+ # Add user1 to group1
+ self.identity_api.add_user_to_group(user1['id'], group1['id'])
+
+ # Now, add grant to user1 and group1 in tenant_bar
+ self.assignment_api.create_grant(user_id=user1['id'],
+ project_id=self.tenant_bar['id'],
+ role_id=self.role_member['id'])
+ self.assignment_api.create_grant(group_id=group1['id'],
+ project_id=self.tenant_bar['id'],
+ role_id=self.role_member['id'])
+
+ # The result is user1 has only one project granted
+ user_projects = self.assignment_api.list_projects_for_user(user1['id'])
+ self.assertThat(user_projects, matchers.HasLength(1))
+
+ # Now, delete user1 grant into tenant_bar and check
+ self.assignment_api.delete_grant(user_id=user1['id'],
+ project_id=self.tenant_bar['id'],
+ role_id=self.role_member['id'])
+
+ # The result is user1 has only one project granted.
+ # Granted through group1.
+ user_projects = self.assignment_api.list_projects_for_user(user1['id'])
+ self.assertThat(user_projects, matchers.HasLength(1))
+
+ def test_list_projects_for_user_with_grants(self):
+ domain = self._get_domain_fixture()
+ new_user = {'name': 'new_user', 'password': uuid.uuid4().hex,
+ 'enabled': True, 'domain_id': domain['id']}
+ new_user = self.identity_api.create_user(new_user)
+
+ group1 = {'name': uuid.uuid4().hex, 'domain_id': domain['id']}
+ group1 = self.identity_api.create_group(group1)
+ group2 = {'name': uuid.uuid4().hex, 'domain_id': domain['id']}
+ group2 = self.identity_api.create_group(group2)
+
+ project1 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
+ 'domain_id': domain['id']}
+ self.resource_api.create_project(project1['id'], project1)
+ project2 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
+ 'domain_id': domain['id']}
+ self.resource_api.create_project(project2['id'], project2)
+
+ self.identity_api.add_user_to_group(new_user['id'],
+ group1['id'])
+ self.identity_api.add_user_to_group(new_user['id'],
+ group2['id'])
+
+ self.assignment_api.create_grant(user_id=new_user['id'],
+ project_id=self.tenant_bar['id'],
+ role_id=self.role_member['id'])
+ self.assignment_api.create_grant(user_id=new_user['id'],
+ project_id=project1['id'],
+ role_id=self.role_admin['id'])
+ self.assignment_api.create_grant(group_id=group2['id'],
+ project_id=project2['id'],
+ role_id=self.role_admin['id'])
+
+ user_projects = self.assignment_api.list_projects_for_user(
+ new_user['id'])
+ self.assertEqual(3, len(user_projects))
+
+ def test_create_duplicate_user_name_in_different_domains(self):
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_create_duplicate_project_name_in_different_domains(self):
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_create_duplicate_group_name_in_different_domains(self):
+ self.skipTest(
+ 'N/A: LDAP does not support multiple domains')
+
+ def test_move_user_between_domains(self):
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_move_user_between_domains_with_clashing_names_fails(self):
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_move_group_between_domains(self):
+ self.skipTest(
+ 'N/A: LDAP does not support multiple domains')
+
+ def test_move_group_between_domains_with_clashing_names_fails(self):
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_move_project_between_domains(self):
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_move_project_between_domains_with_clashing_names_fails(self):
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_get_roles_for_user_and_domain(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_get_roles_for_groups_on_domain(self):
+ self.skipTest('Blocked by bug: 1390125')
+
+ def test_get_roles_for_groups_on_project(self):
+ self.skipTest('Blocked by bug: 1390125')
+
+ def test_list_domains_for_groups(self):
+ self.skipTest('N/A: LDAP does not support multiple domains')
+
+ def test_list_projects_for_groups(self):
+ self.skipTest('Blocked by bug: 1390125')
+
+ def test_domain_delete_hierarchy(self):
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_list_role_assignments_unfiltered(self):
+ new_domain = self._get_domain_fixture()
+ new_user = {'name': uuid.uuid4().hex, 'password': uuid.uuid4().hex,
+ 'enabled': True, 'domain_id': new_domain['id']}
+ new_user = self.identity_api.create_user(new_user)
+ new_group = {'domain_id': new_domain['id'], 'name': uuid.uuid4().hex}
+ new_group = self.identity_api.create_group(new_group)
+ new_project = {'id': uuid.uuid4().hex,
+ 'name': uuid.uuid4().hex,
+ 'domain_id': new_domain['id']}
+ self.resource_api.create_project(new_project['id'], new_project)
+
+ # First check how many role grant already exist
+ existing_assignments = len(self.assignment_api.list_role_assignments())
+
+ self.assignment_api.create_grant(user_id=new_user['id'],
+ project_id=new_project['id'],
+ role_id='other')
+ self.assignment_api.create_grant(group_id=new_group['id'],
+ project_id=new_project['id'],
+ role_id='admin')
+
+ # Read back the list of assignments - check it is gone up by 2
+ after_assignments = len(self.assignment_api.list_role_assignments())
+ self.assertEqual(existing_assignments + 2, after_assignments)
+
+ def test_list_role_assignments_dumb_member(self):
+ self.config_fixture.config(group='ldap', use_dumb_member=True)
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+
+ new_domain = self._get_domain_fixture()
+ new_user = {'name': uuid.uuid4().hex, 'password': uuid.uuid4().hex,
+ 'enabled': True, 'domain_id': new_domain['id']}
+ new_user = self.identity_api.create_user(new_user)
+ new_project = {'id': uuid.uuid4().hex,
+ 'name': uuid.uuid4().hex,
+ 'domain_id': new_domain['id']}
+ self.resource_api.create_project(new_project['id'], new_project)
+ self.assignment_api.create_grant(user_id=new_user['id'],
+ project_id=new_project['id'],
+ role_id='other')
+
+ # Read back the list of assignments and ensure
+ # that the LDAP dumb member isn't listed.
+ assignment_ids = [a['user_id'] for a in
+ self.assignment_api.list_role_assignments()]
+ dumb_id = common_ldap.BaseLdap._dn_to_id(CONF.ldap.dumb_member)
+ self.assertNotIn(dumb_id, assignment_ids)
+
+ def test_list_user_ids_for_project_dumb_member(self):
+ self.config_fixture.config(group='ldap', use_dumb_member=True)
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+
+ user = {'name': uuid.uuid4().hex, 'password': uuid.uuid4().hex,
+ 'enabled': True, 'domain_id': test_backend.DEFAULT_DOMAIN_ID}
+
+ user = self.identity_api.create_user(user)
+ self.assignment_api.add_user_to_project(self.tenant_baz['id'],
+ user['id'])
+ user_ids = self.assignment_api.list_user_ids_for_project(
+ self.tenant_baz['id'])
+
+ self.assertIn(user['id'], user_ids)
+
+ dumb_id = common_ldap.BaseLdap._dn_to_id(CONF.ldap.dumb_member)
+ self.assertNotIn(dumb_id, user_ids)
+
+ def test_multi_group_grants_on_project_domain(self):
+ self.skipTest('Blocked by bug 1101287')
+
+ def test_list_group_members_missing_entry(self):
+ """List group members with deleted user.
+
+ If a group has a deleted entry for a member, the non-deleted members
+ are returned.
+
+ """
+
+ # Create a group
+ group = dict(name=uuid.uuid4().hex,
+ domain_id=CONF.identity.default_domain_id)
+ group_id = self.identity_api.create_group(group)['id']
+
+ # Create a couple of users and add them to the group.
+ user = dict(name=uuid.uuid4().hex,
+ domain_id=CONF.identity.default_domain_id)
+ user_1_id = self.identity_api.create_user(user)['id']
+
+ self.identity_api.add_user_to_group(user_1_id, group_id)
+
+ user = dict(name=uuid.uuid4().hex,
+ domain_id=CONF.identity.default_domain_id)
+ user_2_id = self.identity_api.create_user(user)['id']
+
+ self.identity_api.add_user_to_group(user_2_id, group_id)
+
+ # Delete user 2
+ # NOTE(blk-u): need to go directly to user interface to keep from
+ # updating the group.
+ unused, driver, entity_id = (
+ self.identity_api._get_domain_driver_and_entity_id(user_2_id))
+ driver.user.delete(entity_id)
+
+ # List group users and verify only user 1.
+ res = self.identity_api.list_users_in_group(group_id)
+
+ self.assertEqual(1, len(res), "Expected 1 entry (user_1)")
+ self.assertEqual(user_1_id, res[0]['id'], "Expected user 1 id")
+
+ def test_list_group_members_when_no_members(self):
+ # List group members when there is no member in the group.
+ # No exception should be raised.
+ group = {
+ 'domain_id': CONF.identity.default_domain_id,
+ 'name': uuid.uuid4().hex,
+ 'description': uuid.uuid4().hex}
+ group = self.identity_api.create_group(group)
+
+ # If this doesn't raise, then the test is successful.
+ self.identity_api.list_users_in_group(group['id'])
+
+ def test_list_group_members_dumb_member(self):
+ self.config_fixture.config(group='ldap', use_dumb_member=True)
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+
+ # Create a group
+ group = dict(name=uuid.uuid4().hex,
+ domain_id=CONF.identity.default_domain_id)
+ group_id = self.identity_api.create_group(group)['id']
+
+ # Create a user
+ user = dict(name=uuid.uuid4().hex,
+ domain_id=CONF.identity.default_domain_id)
+ user_id = self.identity_api.create_user(user)['id']
+
+ # Add user to the group
+ self.identity_api.add_user_to_group(user_id, group_id)
+
+ user_ids = self.identity_api.list_users_in_group(group_id)
+ dumb_id = common_ldap.BaseLdap._dn_to_id(CONF.ldap.dumb_member)
+
+ self.assertNotIn(dumb_id, user_ids)
+
+ def test_list_domains(self):
+ domains = self.resource_api.list_domains()
+ self.assertEqual(
+ [resource.calc_default_domain()],
+ domains)
+
+ def test_list_domains_non_default_domain_id(self):
+ # If change the default_domain_id, the ID of the default domain
+ # returned by list_domains changes is the new default_domain_id.
+
+ new_domain_id = uuid.uuid4().hex
+ self.config_fixture.config(group='identity',
+ default_domain_id=new_domain_id)
+
+ domains = self.resource_api.list_domains()
+
+ self.assertEqual(new_domain_id, domains[0]['id'])
+
+ def test_authenticate_requires_simple_bind(self):
+ user = {
+ 'name': 'NO_META',
+ 'domain_id': test_backend.DEFAULT_DOMAIN_ID,
+ 'password': 'no_meta2',
+ 'enabled': True,
+ }
+ user = self.identity_api.create_user(user)
+ self.assignment_api.add_user_to_project(self.tenant_baz['id'],
+ user['id'])
+ driver = self.identity_api._select_identity_driver(
+ user['domain_id'])
+ driver.user.LDAP_USER = None
+ driver.user.LDAP_PASSWORD = None
+
+ self.assertRaises(AssertionError,
+ self.identity_api.authenticate,
+ context={},
+ user_id=user['id'],
+ password=None)
+
+ # (spzala)The group and domain crud tests below override the standard ones
+ # in test_backend.py so that we can exclude the update name test, since we
+ # do not yet support the update of either group or domain names with LDAP.
+ # In the tests below, the update is demonstrated by updating description.
+ # Refer to bug 1136403 for more detail.
+ def test_group_crud(self):
+ group = {
+ 'domain_id': CONF.identity.default_domain_id,
+ 'name': uuid.uuid4().hex,
+ 'description': uuid.uuid4().hex}
+ group = self.identity_api.create_group(group)
+ group_ref = self.identity_api.get_group(group['id'])
+ self.assertDictEqual(group_ref, group)
+ group['description'] = uuid.uuid4().hex
+ self.identity_api.update_group(group['id'], group)
+ group_ref = self.identity_api.get_group(group['id'])
+ self.assertDictEqual(group_ref, group)
+
+ self.identity_api.delete_group(group['id'])
+ self.assertRaises(exception.GroupNotFound,
+ self.identity_api.get_group,
+ group['id'])
+
+ @tests.skip_if_cache_disabled('identity')
+ def test_cache_layer_group_crud(self):
+ group = {
+ 'domain_id': CONF.identity.default_domain_id,
+ 'name': uuid.uuid4().hex}
+ group = self.identity_api.create_group(group)
+ # cache the result
+ group_ref = self.identity_api.get_group(group['id'])
+ # delete the group bypassing identity api.
+ domain_id, driver, entity_id = (
+ self.identity_api._get_domain_driver_and_entity_id(group['id']))
+ driver.delete_group(entity_id)
+
+ self.assertEqual(group_ref,
+ self.identity_api.get_group(group['id']))
+ self.identity_api.get_group.invalidate(self.identity_api, group['id'])
+ self.assertRaises(exception.GroupNotFound,
+ self.identity_api.get_group, group['id'])
+
+ group = {
+ 'domain_id': CONF.identity.default_domain_id,
+ 'name': uuid.uuid4().hex}
+ group = self.identity_api.create_group(group)
+ # cache the result
+ self.identity_api.get_group(group['id'])
+ group['description'] = uuid.uuid4().hex
+ group_ref = self.identity_api.update_group(group['id'], group)
+ self.assertDictContainsSubset(self.identity_api.get_group(group['id']),
+ group_ref)
+
+ def test_create_user_none_mapping(self):
+ # When create a user where an attribute maps to None, the entry is
+ # created without that attribute and it doesn't fail with a TypeError.
+ conf = self.get_config(CONF.identity.default_domain_id)
+ conf.ldap.user_attribute_ignore = ['enabled', 'email',
+ 'tenants', 'tenantId']
+ self.reload_backends(CONF.identity.default_domain_id)
+
+ user = {'name': u'fäké1',
+ 'password': u'fäképass1',
+ 'domain_id': CONF.identity.default_domain_id,
+ 'default_project_id': 'maps_to_none',
+ }
+
+ # If this doesn't raise, then the test is successful.
+ user = self.identity_api.create_user(user)
+
+ def test_create_user_with_boolean_string_names(self):
+ # Ensure that any attribute that is equal to the string 'TRUE'
+ # or 'FALSE' will not be converted to a boolean value, it
+ # should be returned as is.
+ boolean_strings = ['TRUE', 'FALSE', 'true', 'false', 'True', 'False',
+ 'TrUe' 'FaLse']
+ for name in boolean_strings:
+ user = {
+ 'name': name,
+ 'domain_id': CONF.identity.default_domain_id}
+ user_ref = self.identity_api.create_user(user)
+ user_info = self.identity_api.get_user(user_ref['id'])
+ self.assertEqual(name, user_info['name'])
+ # Delete the user to ensure that the Keystone uniqueness
+ # requirements combined with the case-insensitive nature of a
+ # typical LDAP schema does not cause subsequent names in
+ # boolean_strings to clash.
+ self.identity_api.delete_user(user_ref['id'])
+
+ def test_unignored_user_none_mapping(self):
+ # Ensure that an attribute that maps to None that is not explicitly
+ # ignored in configuration is implicitly ignored without triggering
+ # an error.
+ conf = self.get_config(CONF.identity.default_domain_id)
+ conf.ldap.user_attribute_ignore = ['enabled', 'email',
+ 'tenants', 'tenantId']
+ self.reload_backends(CONF.identity.default_domain_id)
+
+ user = {'name': u'fäké1',
+ 'password': u'fäképass1',
+ 'domain_id': CONF.identity.default_domain_id,
+ }
+
+ user_ref = self.identity_api.create_user(user)
+
+ # If this doesn't raise, then the test is successful.
+ self.identity_api.get_user(user_ref['id'])
+
+ def test_update_user_name(self):
+ """A user's name cannot be changed through the LDAP driver."""
+ self.assertRaises(exception.Conflict,
+ super(BaseLDAPIdentity, self).test_update_user_name)
+
+ def test_arbitrary_attributes_are_returned_from_get_user(self):
+ self.skipTest("Using arbitrary attributes doesn't work under LDAP")
+
+ def test_new_arbitrary_attributes_are_returned_from_update_user(self):
+ self.skipTest("Using arbitrary attributes doesn't work under LDAP")
+
+ def test_updated_arbitrary_attributes_are_returned_from_update_user(self):
+ self.skipTest("Using arbitrary attributes doesn't work under LDAP")
+
+ def test_cache_layer_domain_crud(self):
+ # TODO(morganfainberg): This also needs to be removed when full LDAP
+ # implementation is submitted. No need to duplicate the above test,
+ # just skip this time.
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_user_id_comma(self):
+ """Even if the user has a , in their ID, groups can be listed."""
+
+ # Create a user with a , in their ID
+ # NOTE(blk-u): the DN for this user is hard-coded in fakeldap!
+
+ # Since we want to fake up this special ID, we'll squirt this
+ # direct into the driver and bypass the manager layer.
+ user_id = u'Doe, John'
+ user = {
+ 'id': user_id,
+ 'name': self.getUniqueString(),
+ 'password': self.getUniqueString(),
+ 'domain_id': CONF.identity.default_domain_id,
+ }
+ user = self.identity_api.driver.create_user(user_id, user)
+
+ # Now we'll use the manager to discover it, which will create a
+ # Public ID for it.
+ ref_list = self.identity_api.list_users()
+ public_user_id = None
+ for ref in ref_list:
+ if ref['name'] == user['name']:
+ public_user_id = ref['id']
+ break
+
+ # Create a group
+ group_id = uuid.uuid4().hex
+ group = {
+ 'id': group_id,
+ 'name': self.getUniqueString(prefix='tuidc'),
+ 'description': self.getUniqueString(),
+ 'domain_id': CONF.identity.default_domain_id,
+ }
+ group = self.identity_api.driver.create_group(group_id, group)
+ # Now we'll use the manager to discover it, which will create a
+ # Public ID for it.
+ ref_list = self.identity_api.list_groups()
+ public_group_id = None
+ for ref in ref_list:
+ if ref['name'] == group['name']:
+ public_group_id = ref['id']
+ break
+
+ # Put the user in the group
+ self.identity_api.add_user_to_group(public_user_id, public_group_id)
+
+ # List groups for user.
+ ref_list = self.identity_api.list_groups_for_user(public_user_id)
+
+ group['id'] = public_group_id
+ self.assertThat(ref_list, matchers.Equals([group]))
+
+ def test_user_id_comma_grants(self):
+ """Even if the user has a , in their ID, can get user and group grants.
+ """
+
+ # Create a user with a , in their ID
+ # NOTE(blk-u): the DN for this user is hard-coded in fakeldap!
+
+ # Since we want to fake up this special ID, we'll squirt this
+ # direct into the driver and bypass the manager layer
+ user_id = u'Doe, John'
+ user = {
+ 'id': user_id,
+ 'name': self.getUniqueString(),
+ 'password': self.getUniqueString(),
+ 'domain_id': CONF.identity.default_domain_id,
+ }
+ self.identity_api.driver.create_user(user_id, user)
+
+ # Now we'll use the manager to discover it, which will create a
+ # Public ID for it.
+ ref_list = self.identity_api.list_users()
+ public_user_id = None
+ for ref in ref_list:
+ if ref['name'] == user['name']:
+ public_user_id = ref['id']
+ break
+
+ # Grant the user a role on a project.
+
+ role_id = 'member'
+ project_id = self.tenant_baz['id']
+
+ self.assignment_api.create_grant(role_id, user_id=public_user_id,
+ project_id=project_id)
+
+ role_ref = self.assignment_api.get_grant(role_id,
+ user_id=public_user_id,
+ project_id=project_id)
+
+ self.assertEqual(role_id, role_ref['id'])
+
+ def test_user_enabled_ignored_disable_error(self):
+ # When the server is configured so that the enabled attribute is
+ # ignored for users, users cannot be disabled.
+
+ self.config_fixture.config(group='ldap',
+ user_attribute_ignore=['enabled'])
+
+ # Need to re-load backends for the config change to take effect.
+ self.load_backends()
+
+ # Attempt to disable the user.
+ self.assertRaises(exception.ForbiddenAction,
+ self.identity_api.update_user, self.user_foo['id'],
+ {'enabled': False})
+
+ user_info = self.identity_api.get_user(self.user_foo['id'])
+
+ # If 'enabled' is ignored then 'enabled' isn't returned as part of the
+ # ref.
+ self.assertNotIn('enabled', user_info)
+
+ def test_group_enabled_ignored_disable_error(self):
+ # When the server is configured so that the enabled attribute is
+ # ignored for groups, groups cannot be disabled.
+
+ self.config_fixture.config(group='ldap',
+ group_attribute_ignore=['enabled'])
+
+ # Need to re-load backends for the config change to take effect.
+ self.load_backends()
+
+ # There's no group fixture so create a group.
+ new_domain = self._get_domain_fixture()
+ new_group = {'domain_id': new_domain['id'],
+ 'name': uuid.uuid4().hex}
+ new_group = self.identity_api.create_group(new_group)
+
+ # Attempt to disable the group.
+ self.assertRaises(exception.ForbiddenAction,
+ self.identity_api.update_group, new_group['id'],
+ {'enabled': False})
+
+ group_info = self.identity_api.get_group(new_group['id'])
+
+ # If 'enabled' is ignored then 'enabled' isn't returned as part of the
+ # ref.
+ self.assertNotIn('enabled', group_info)
+
+ def test_project_enabled_ignored_disable_error(self):
+ # When the server is configured so that the enabled attribute is
+ # ignored for projects, projects cannot be disabled.
+
+ self.config_fixture.config(group='ldap',
+ project_attribute_ignore=['enabled'])
+
+ # Need to re-load backends for the config change to take effect.
+ self.load_backends()
+
+ # Attempt to disable the project.
+ self.assertRaises(exception.ForbiddenAction,
+ self.resource_api.update_project,
+ self.tenant_baz['id'], {'enabled': False})
+
+ project_info = self.resource_api.get_project(self.tenant_baz['id'])
+
+ # Unlike other entities, if 'enabled' is ignored then 'enabled' is
+ # returned as part of the ref.
+ self.assertIs(True, project_info['enabled'])
+
+
+class LDAPIdentity(BaseLDAPIdentity, tests.TestCase):
+
+ def setUp(self):
+ # NOTE(dstanek): The database must be setup prior to calling the
+ # parent's setUp. The parent's setUp uses services (like
+ # credentials) that require a database.
+ self.useFixture(database.Database())
+ super(LDAPIdentity, self).setUp()
+
+ def load_fixtures(self, fixtures):
+ # Override super impl since need to create group container.
+ create_group_container(self.identity_api)
+ super(LDAPIdentity, self).load_fixtures(fixtures)
+
+ def test_configurable_allowed_project_actions(self):
+ tenant = {'id': u'fäké1', 'name': u'fäké1', 'enabled': True}
+ self.resource_api.create_project(u'fäké1', tenant)
+ tenant_ref = self.resource_api.get_project(u'fäké1')
+ self.assertEqual(u'fäké1', tenant_ref['id'])
+
+ tenant['enabled'] = False
+ self.resource_api.update_project(u'fäké1', tenant)
+
+ self.resource_api.delete_project(u'fäké1')
+ self.assertRaises(exception.ProjectNotFound,
+ self.resource_api.get_project,
+ u'fäké1')
+
+ def test_configurable_subtree_delete(self):
+ self.config_fixture.config(group='ldap', allow_subtree_delete=True)
+ self.load_backends()
+
+ project1 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id}
+ self.resource_api.create_project(project1['id'], project1)
+
+ role1 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
+ self.role_api.create_role(role1['id'], role1)
+
+ user1 = {'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id,
+ 'password': uuid.uuid4().hex,
+ 'enabled': True}
+ user1 = self.identity_api.create_user(user1)
+
+ self.assignment_api.add_role_to_user_and_project(
+ user_id=user1['id'],
+ tenant_id=project1['id'],
+ role_id=role1['id'])
+
+ self.resource_api.delete_project(project1['id'])
+ self.assertRaises(exception.ProjectNotFound,
+ self.resource_api.get_project,
+ project1['id'])
+
+ self.resource_api.create_project(project1['id'], project1)
+
+ list = self.assignment_api.get_roles_for_user_and_project(
+ user1['id'],
+ project1['id'])
+ self.assertEqual(0, len(list))
+
+ def test_configurable_forbidden_project_actions(self):
+ self.config_fixture.config(
+ group='ldap', project_allow_create=False,
+ project_allow_update=False, project_allow_delete=False)
+ self.load_backends()
+
+ tenant = {'id': u'fäké1', 'name': u'fäké1'}
+ self.assertRaises(exception.ForbiddenAction,
+ self.resource_api.create_project,
+ u'fäké1',
+ tenant)
+
+ self.tenant_bar['enabled'] = False
+ self.assertRaises(exception.ForbiddenAction,
+ self.resource_api.update_project,
+ self.tenant_bar['id'],
+ self.tenant_bar)
+ self.assertRaises(exception.ForbiddenAction,
+ self.resource_api.delete_project,
+ self.tenant_bar['id'])
+
+ def test_project_filter(self):
+ tenant_ref = self.resource_api.get_project(self.tenant_bar['id'])
+ self.assertDictEqual(tenant_ref, self.tenant_bar)
+
+ self.config_fixture.config(group='ldap',
+ project_filter='(CN=DOES_NOT_MATCH)')
+ self.load_backends()
+ # NOTE(morganfainberg): CONF.ldap.project_filter will not be
+ # dynamically changed at runtime. This invalidate is a work-around for
+ # the expectation that it is safe to change config values in tests that
+ # could affect what the drivers would return up to the manager. This
+ # solves this assumption when working with aggressive (on-create)
+ # cache population.
+ self.role_api.get_role.invalidate(self.role_api,
+ self.role_member['id'])
+ self.role_api.get_role(self.role_member['id'])
+ self.resource_api.get_project.invalidate(self.resource_api,
+ self.tenant_bar['id'])
+ self.assertRaises(exception.ProjectNotFound,
+ self.resource_api.get_project,
+ self.tenant_bar['id'])
+
+ def test_dumb_member(self):
+ self.config_fixture.config(group='ldap', use_dumb_member=True)
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+ dumb_id = common_ldap.BaseLdap._dn_to_id(CONF.ldap.dumb_member)
+ self.assertRaises(exception.UserNotFound,
+ self.identity_api.get_user,
+ dumb_id)
+
+ def test_project_attribute_mapping(self):
+ self.config_fixture.config(
+ group='ldap', project_name_attribute='ou',
+ project_desc_attribute='description',
+ project_enabled_attribute='enabled')
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+ # NOTE(morganfainberg): CONF.ldap.project_name_attribute,
+ # CONF.ldap.project_desc_attribute, and
+ # CONF.ldap.project_enabled_attribute will not be
+ # dynamically changed at runtime. This invalidate is a work-around for
+ # the expectation that it is safe to change config values in tests that
+ # could affect what the drivers would return up to the manager. This
+ # solves this assumption when working with aggressive (on-create)
+ # cache population.
+ self.resource_api.get_project.invalidate(self.resource_api,
+ self.tenant_baz['id'])
+ tenant_ref = self.resource_api.get_project(self.tenant_baz['id'])
+ self.assertEqual(self.tenant_baz['id'], tenant_ref['id'])
+ self.assertEqual(self.tenant_baz['name'], tenant_ref['name'])
+ self.assertEqual(
+ self.tenant_baz['description'],
+ tenant_ref['description'])
+ self.assertEqual(self.tenant_baz['enabled'], tenant_ref['enabled'])
+
+ self.config_fixture.config(group='ldap',
+ project_name_attribute='description',
+ project_desc_attribute='ou')
+ self.load_backends()
+ # NOTE(morganfainberg): CONF.ldap.project_name_attribute,
+ # CONF.ldap.project_desc_attribute, and
+ # CONF.ldap.project_enabled_attribute will not be
+ # dynamically changed at runtime. This invalidate is a work-around for
+ # the expectation that it is safe to change config values in tests that
+ # could affect what the drivers would return up to the manager. This
+ # solves this assumption when working with aggressive (on-create)
+ # cache population.
+ self.resource_api.get_project.invalidate(self.resource_api,
+ self.tenant_baz['id'])
+ tenant_ref = self.resource_api.get_project(self.tenant_baz['id'])
+ self.assertEqual(self.tenant_baz['id'], tenant_ref['id'])
+ self.assertEqual(self.tenant_baz['description'], tenant_ref['name'])
+ self.assertEqual(self.tenant_baz['name'], tenant_ref['description'])
+ self.assertEqual(self.tenant_baz['enabled'], tenant_ref['enabled'])
+
+ def test_project_attribute_ignore(self):
+ self.config_fixture.config(
+ group='ldap',
+ project_attribute_ignore=['name', 'description', 'enabled'])
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+ # NOTE(morganfainberg): CONF.ldap.project_attribute_ignore will not be
+ # dynamically changed at runtime. This invalidate is a work-around for
+ # the expectation that it is safe to change configs values in tests
+ # that could affect what the drivers would return up to the manager.
+ # This solves this assumption when working with aggressive (on-create)
+ # cache population.
+ self.resource_api.get_project.invalidate(self.resource_api,
+ self.tenant_baz['id'])
+ tenant_ref = self.resource_api.get_project(self.tenant_baz['id'])
+ self.assertEqual(self.tenant_baz['id'], tenant_ref['id'])
+ self.assertNotIn('name', tenant_ref)
+ self.assertNotIn('description', tenant_ref)
+ self.assertNotIn('enabled', tenant_ref)
+
+ def test_user_enable_attribute_mask(self):
+ self.config_fixture.config(group='ldap', user_enabled_mask=2,
+ user_enabled_default='512')
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+
+ user = {'name': u'fäké1', 'enabled': True,
+ 'domain_id': CONF.identity.default_domain_id}
+
+ user_ref = self.identity_api.create_user(user)
+
+ # Use assertIs rather than assertTrue because assertIs will assert the
+ # value is a Boolean as expected.
+ self.assertIs(user_ref['enabled'], True)
+ self.assertNotIn('enabled_nomask', user_ref)
+
+ enabled_vals = self.get_user_enabled_vals(user_ref)
+ self.assertEqual([512], enabled_vals)
+
+ user_ref = self.identity_api.get_user(user_ref['id'])
+ self.assertIs(user_ref['enabled'], True)
+ self.assertNotIn('enabled_nomask', user_ref)
+
+ user['enabled'] = False
+ user_ref = self.identity_api.update_user(user_ref['id'], user)
+ self.assertIs(user_ref['enabled'], False)
+ self.assertNotIn('enabled_nomask', user_ref)
+
+ enabled_vals = self.get_user_enabled_vals(user_ref)
+ self.assertEqual([514], enabled_vals)
+
+ user_ref = self.identity_api.get_user(user_ref['id'])
+ self.assertIs(user_ref['enabled'], False)
+ self.assertNotIn('enabled_nomask', user_ref)
+
+ user['enabled'] = True
+ user_ref = self.identity_api.update_user(user_ref['id'], user)
+ self.assertIs(user_ref['enabled'], True)
+ self.assertNotIn('enabled_nomask', user_ref)
+
+ enabled_vals = self.get_user_enabled_vals(user_ref)
+ self.assertEqual([512], enabled_vals)
+
+ user_ref = self.identity_api.get_user(user_ref['id'])
+ self.assertIs(user_ref['enabled'], True)
+ self.assertNotIn('enabled_nomask', user_ref)
+
+ def test_user_enabled_invert(self):
+ self.config_fixture.config(group='ldap', user_enabled_invert=True,
+ user_enabled_default=False)
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+
+ user1 = {'name': u'fäké1', 'enabled': True,
+ 'domain_id': CONF.identity.default_domain_id}
+
+ user2 = {'name': u'fäké2', 'enabled': False,
+ 'domain_id': CONF.identity.default_domain_id}
+
+ user3 = {'name': u'fäké3',
+ 'domain_id': CONF.identity.default_domain_id}
+
+ # Ensure that the LDAP attribute is False for a newly created
+ # enabled user.
+ user_ref = self.identity_api.create_user(user1)
+ self.assertIs(True, user_ref['enabled'])
+ enabled_vals = self.get_user_enabled_vals(user_ref)
+ self.assertEqual([False], enabled_vals)
+ user_ref = self.identity_api.get_user(user_ref['id'])
+ self.assertIs(True, user_ref['enabled'])
+
+ # Ensure that the LDAP attribute is True for a disabled user.
+ user1['enabled'] = False
+ user_ref = self.identity_api.update_user(user_ref['id'], user1)
+ self.assertIs(False, user_ref['enabled'])
+ enabled_vals = self.get_user_enabled_vals(user_ref)
+ self.assertEqual([True], enabled_vals)
+
+ # Enable the user and ensure that the LDAP attribute is True again.
+ user1['enabled'] = True
+ user_ref = self.identity_api.update_user(user_ref['id'], user1)
+ self.assertIs(True, user_ref['enabled'])
+ enabled_vals = self.get_user_enabled_vals(user_ref)
+ self.assertEqual([False], enabled_vals)
+
+ # Ensure that the LDAP attribute is True for a newly created
+ # disabled user.
+ user_ref = self.identity_api.create_user(user2)
+ self.assertIs(False, user_ref['enabled'])
+ enabled_vals = self.get_user_enabled_vals(user_ref)
+ self.assertEqual([True], enabled_vals)
+ user_ref = self.identity_api.get_user(user_ref['id'])
+ self.assertIs(False, user_ref['enabled'])
+
+ # Ensure that the LDAP attribute is inverted for a newly created
+ # user when the user_enabled_default setting is used.
+ user_ref = self.identity_api.create_user(user3)
+ self.assertIs(True, user_ref['enabled'])
+ enabled_vals = self.get_user_enabled_vals(user_ref)
+ self.assertEqual([False], enabled_vals)
+ user_ref = self.identity_api.get_user(user_ref['id'])
+ self.assertIs(True, user_ref['enabled'])
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_user_enabled_invert_no_enabled_value(self, mock_ldap_get):
+ self.config_fixture.config(group='ldap', user_enabled_invert=True,
+ user_enabled_default=False)
+ # Mock the search results to return an entry with
+ # no enabled value.
+ mock_ldap_get.return_value = (
+ 'cn=junk,dc=example,dc=com',
+ {
+ 'sn': [uuid.uuid4().hex],
+ 'email': [uuid.uuid4().hex],
+ 'cn': ['junk']
+ }
+ )
+
+ user_api = identity.backends.ldap.UserApi(CONF)
+ user_ref = user_api.get('junk')
+ # Ensure that the model enabled attribute is inverted
+ # from the resource default.
+ self.assertIs(not CONF.ldap.user_enabled_default, user_ref['enabled'])
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_user_enabled_invert_default_str_value(self, mock_ldap_get):
+ self.config_fixture.config(group='ldap', user_enabled_invert=True,
+ user_enabled_default='False')
+ # Mock the search results to return an entry with
+ # no enabled value.
+ mock_ldap_get.return_value = (
+ 'cn=junk,dc=example,dc=com',
+ {
+ 'sn': [uuid.uuid4().hex],
+ 'email': [uuid.uuid4().hex],
+ 'cn': ['junk']
+ }
+ )
+
+ user_api = identity.backends.ldap.UserApi(CONF)
+ user_ref = user_api.get('junk')
+ # Ensure that the model enabled attribute is inverted
+ # from the resource default.
+ self.assertIs(True, user_ref['enabled'])
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_user_enabled_attribute_handles_expired(self, mock_ldap_get):
+ # If using 'passwordisexpired' as enabled attribute, and inverting it,
+ # Then an unauthorized user (expired password) should not be enabled.
+ self.config_fixture.config(group='ldap', user_enabled_invert=True,
+ user_enabled_attribute='passwordisexpired')
+ mock_ldap_get.return_value = (
+ u'uid=123456789,c=us,ou=our_ldap,o=acme.com',
+ {
+ 'uid': [123456789],
+ 'mail': ['shaun@acme.com'],
+ 'passwordisexpired': ['TRUE'],
+ 'cn': ['uid=123456789,c=us,ou=our_ldap,o=acme.com']
+ }
+ )
+
+ user_api = identity.backends.ldap.UserApi(CONF)
+ user_ref = user_api.get('123456789')
+ self.assertIs(False, user_ref['enabled'])
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_user_enabled_attribute_handles_utf8(self, mock_ldap_get):
+ # If using 'passwordisexpired' as enabled attribute, and inverting it,
+ # and the result is utf8 encoded, then the an authorized user should
+ # be enabled.
+ self.config_fixture.config(group='ldap', user_enabled_invert=True,
+ user_enabled_attribute='passwordisexpired')
+ mock_ldap_get.return_value = (
+ u'uid=123456789,c=us,ou=our_ldap,o=acme.com',
+ {
+ 'uid': [123456789],
+ 'mail': [u'shaun@acme.com'],
+ 'passwordisexpired': [u'false'],
+ 'cn': [u'uid=123456789,c=us,ou=our_ldap,o=acme.com']
+ }
+ )
+
+ user_api = identity.backends.ldap.UserApi(CONF)
+ user_ref = user_api.get('123456789')
+ self.assertIs(True, user_ref['enabled'])
+
+ @mock.patch.object(common_ldap_core.KeystoneLDAPHandler, 'simple_bind_s')
+ def test_user_api_get_connection_no_user_password(self, mocked_method):
+ """Don't bind in case the user and password are blank."""
+ # Ensure the username/password are in-fact blank
+ self.config_fixture.config(group='ldap', user=None, password=None)
+ user_api = identity.backends.ldap.UserApi(CONF)
+ user_api.get_connection(user=None, password=None)
+ self.assertFalse(mocked_method.called,
+ msg='`simple_bind_s` method was unexpectedly called')
+
+ @mock.patch.object(common_ldap_core.KeystoneLDAPHandler, 'connect')
+ def test_chase_referrals_off(self, mocked_fakeldap):
+ self.config_fixture.config(
+ group='ldap',
+ url='fake://memory',
+ chase_referrals=False)
+ user_api = identity.backends.ldap.UserApi(CONF)
+ user_api.get_connection(user=None, password=None)
+
+ # The last call_arg should be a dictionary and should contain
+ # chase_referrals. Check to make sure the value of chase_referrals
+ # is as expected.
+ self.assertFalse(mocked_fakeldap.call_args[-1]['chase_referrals'])
+
+ @mock.patch.object(common_ldap_core.KeystoneLDAPHandler, 'connect')
+ def test_chase_referrals_on(self, mocked_fakeldap):
+ self.config_fixture.config(
+ group='ldap',
+ url='fake://memory',
+ chase_referrals=True)
+ user_api = identity.backends.ldap.UserApi(CONF)
+ user_api.get_connection(user=None, password=None)
+
+ # The last call_arg should be a dictionary and should contain
+ # chase_referrals. Check to make sure the value of chase_referrals
+ # is as expected.
+ self.assertTrue(mocked_fakeldap.call_args[-1]['chase_referrals'])
+
+ @mock.patch.object(common_ldap_core.KeystoneLDAPHandler, 'connect')
+ def test_debug_level_set(self, mocked_fakeldap):
+ level = 12345
+ self.config_fixture.config(
+ group='ldap',
+ url='fake://memory',
+ debug_level=level)
+ user_api = identity.backends.ldap.UserApi(CONF)
+ user_api.get_connection(user=None, password=None)
+
+ # The last call_arg should be a dictionary and should contain
+ # debug_level. Check to make sure the value of debug_level
+ # is as expected.
+ self.assertEqual(level, mocked_fakeldap.call_args[-1]['debug_level'])
+
+ def test_wrong_ldap_scope(self):
+ self.config_fixture.config(group='ldap', query_scope=uuid.uuid4().hex)
+ self.assertRaisesRegexp(
+ ValueError,
+ 'Invalid LDAP scope: %s. *' % CONF.ldap.query_scope,
+ identity.backends.ldap.Identity)
+
+ def test_wrong_alias_dereferencing(self):
+ self.config_fixture.config(group='ldap',
+ alias_dereferencing=uuid.uuid4().hex)
+ self.assertRaisesRegexp(
+ ValueError,
+ 'Invalid LDAP deref option: %s\.' % CONF.ldap.alias_dereferencing,
+ identity.backends.ldap.Identity)
+
+ def test_is_dumb_member(self):
+ self.config_fixture.config(group='ldap',
+ use_dumb_member=True)
+ self.load_backends()
+
+ dn = 'cn=dumb,dc=nonexistent'
+ self.assertTrue(self.identity_api.driver.user._is_dumb_member(dn))
+
+ def test_is_dumb_member_upper_case_keys(self):
+ self.config_fixture.config(group='ldap',
+ use_dumb_member=True)
+ self.load_backends()
+
+ dn = 'CN=dumb,DC=nonexistent'
+ self.assertTrue(self.identity_api.driver.user._is_dumb_member(dn))
+
+ def test_is_dumb_member_with_false_use_dumb_member(self):
+ self.config_fixture.config(group='ldap',
+ use_dumb_member=False)
+ self.load_backends()
+ dn = 'cn=dumb,dc=nonexistent'
+ self.assertFalse(self.identity_api.driver.user._is_dumb_member(dn))
+
+ def test_is_dumb_member_not_dumb(self):
+ self.config_fixture.config(group='ldap',
+ use_dumb_member=True)
+ self.load_backends()
+ dn = 'ou=some,dc=example.com'
+ self.assertFalse(self.identity_api.driver.user._is_dumb_member(dn))
+
+ def test_user_extra_attribute_mapping(self):
+ self.config_fixture.config(
+ group='ldap',
+ user_additional_attribute_mapping=['description:name'])
+ self.load_backends()
+ user = {
+ 'name': 'EXTRA_ATTRIBUTES',
+ 'password': 'extra',
+ 'domain_id': CONF.identity.default_domain_id
+ }
+ user = self.identity_api.create_user(user)
+ dn, attrs = self.identity_api.driver.user._ldap_get(user['id'])
+ self.assertThat([user['name']], matchers.Equals(attrs['description']))
+
+ def test_user_extra_attribute_mapping_description_is_returned(self):
+ # Given a mapping like description:description, the description is
+ # returned.
+
+ self.config_fixture.config(
+ group='ldap',
+ user_additional_attribute_mapping=['description:description'])
+ self.load_backends()
+
+ description = uuid.uuid4().hex
+ user = {
+ 'name': uuid.uuid4().hex,
+ 'description': description,
+ 'password': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id
+ }
+ user = self.identity_api.create_user(user)
+ res = self.identity_api.driver.user.get_all()
+
+ new_user = [u for u in res if u['id'] == user['id']][0]
+ self.assertThat(new_user['description'], matchers.Equals(description))
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_user_mixed_case_attribute(self, mock_ldap_get):
+ # Mock the search results to return attribute names
+ # with unexpected case.
+ mock_ldap_get.return_value = (
+ 'cn=junk,dc=example,dc=com',
+ {
+ 'sN': [uuid.uuid4().hex],
+ 'MaIl': [uuid.uuid4().hex],
+ 'cn': ['junk']
+ }
+ )
+ user = self.identity_api.get_user('junk')
+ self.assertEqual(mock_ldap_get.return_value[1]['sN'][0],
+ user['name'])
+ self.assertEqual(mock_ldap_get.return_value[1]['MaIl'][0],
+ user['email'])
+
+ def test_parse_extra_attribute_mapping(self):
+ option_list = ['description:name', 'gecos:password',
+ 'fake:invalid', 'invalid1', 'invalid2:',
+ 'description:name:something']
+ mapping = self.identity_api.driver.user._parse_extra_attrs(option_list)
+ expected_dict = {'description': 'name', 'gecos': 'password',
+ 'fake': 'invalid', 'invalid2': ''}
+ self.assertDictEqual(expected_dict, mapping)
+
+# TODO(henry-nash): These need to be removed when the full LDAP implementation
+# is submitted - see Bugs 1092187, 1101287, 1101276, 1101289
+
+ def test_domain_crud(self):
+ domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
+ 'enabled': True, 'description': uuid.uuid4().hex}
+ self.assertRaises(exception.Forbidden,
+ self.resource_api.create_domain,
+ domain['id'],
+ domain)
+ self.assertRaises(exception.Conflict,
+ self.resource_api.create_domain,
+ CONF.identity.default_domain_id,
+ domain)
+ self.assertRaises(exception.DomainNotFound,
+ self.resource_api.get_domain,
+ domain['id'])
+
+ domain['description'] = uuid.uuid4().hex
+ self.assertRaises(exception.DomainNotFound,
+ self.resource_api.update_domain,
+ domain['id'],
+ domain)
+ self.assertRaises(exception.Forbidden,
+ self.resource_api.update_domain,
+ CONF.identity.default_domain_id,
+ domain)
+ self.assertRaises(exception.DomainNotFound,
+ self.resource_api.get_domain,
+ domain['id'])
+ self.assertRaises(exception.DomainNotFound,
+ self.resource_api.delete_domain,
+ domain['id'])
+ self.assertRaises(exception.Forbidden,
+ self.resource_api.delete_domain,
+ CONF.identity.default_domain_id)
+ self.assertRaises(exception.DomainNotFound,
+ self.resource_api.get_domain,
+ domain['id'])
+
+ @tests.skip_if_no_multiple_domains_support
+ def test_create_domain_case_sensitivity(self):
+ # domains are read-only, so case sensitivity isn't an issue
+ ref = {
+ 'id': uuid.uuid4().hex,
+ 'name': uuid.uuid4().hex}
+ self.assertRaises(exception.Forbidden,
+ self.resource_api.create_domain,
+ ref['id'],
+ ref)
+
+ def test_cache_layer_domain_crud(self):
+ # TODO(morganfainberg): This also needs to be removed when full LDAP
+ # implementation is submitted. No need to duplicate the above test,
+ # just skip this time.
+ self.skipTest('Domains are read-only against LDAP')
+
+ def test_domain_rename_invalidates_get_domain_by_name_cache(self):
+ parent = super(LDAPIdentity, self)
+ self.assertRaises(
+ exception.Forbidden,
+ parent.test_domain_rename_invalidates_get_domain_by_name_cache)
+
+ def test_project_rename_invalidates_get_project_by_name_cache(self):
+ parent = super(LDAPIdentity, self)
+ self.assertRaises(
+ exception.Forbidden,
+ parent.test_project_rename_invalidates_get_project_by_name_cache)
+
+ def test_project_crud(self):
+ # NOTE(topol): LDAP implementation does not currently support the
+ # updating of a project name so this method override
+ # provides a different update test
+ project = {'id': uuid.uuid4().hex,
+ 'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id,
+ 'description': uuid.uuid4().hex,
+ 'enabled': True,
+ 'parent_id': None}
+ self.resource_api.create_project(project['id'], project)
+ project_ref = self.resource_api.get_project(project['id'])
+
+ self.assertDictEqual(project_ref, project)
+
+ project['description'] = uuid.uuid4().hex
+ self.resource_api.update_project(project['id'], project)
+ project_ref = self.resource_api.get_project(project['id'])
+ self.assertDictEqual(project_ref, project)
+
+ self.resource_api.delete_project(project['id'])
+ self.assertRaises(exception.ProjectNotFound,
+ self.resource_api.get_project,
+ project['id'])
+
+ @tests.skip_if_cache_disabled('assignment')
+ def test_cache_layer_project_crud(self):
+ # NOTE(morganfainberg): LDAP implementation does not currently support
+ # updating project names. This method override provides a different
+ # update test.
+ project = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id,
+ 'description': uuid.uuid4().hex}
+ project_id = project['id']
+ # Create a project
+ self.resource_api.create_project(project_id, project)
+ self.resource_api.get_project(project_id)
+ updated_project = copy.deepcopy(project)
+ updated_project['description'] = uuid.uuid4().hex
+ # Update project, bypassing resource manager
+ self.resource_api.driver.update_project(project_id,
+ updated_project)
+ # Verify get_project still returns the original project_ref
+ self.assertDictContainsSubset(
+ project, self.resource_api.get_project(project_id))
+ # Invalidate cache
+ self.resource_api.get_project.invalidate(self.resource_api,
+ project_id)
+ # Verify get_project now returns the new project
+ self.assertDictContainsSubset(
+ updated_project,
+ self.resource_api.get_project(project_id))
+ # Update project using the resource_api manager back to original
+ self.resource_api.update_project(project['id'], project)
+ # Verify get_project returns the original project_ref
+ self.assertDictContainsSubset(
+ project, self.resource_api.get_project(project_id))
+ # Delete project bypassing resource_api
+ self.resource_api.driver.delete_project(project_id)
+ # Verify get_project still returns the project_ref
+ self.assertDictContainsSubset(
+ project, self.resource_api.get_project(project_id))
+ # Invalidate cache
+ self.resource_api.get_project.invalidate(self.resource_api,
+ project_id)
+ # Verify ProjectNotFound now raised
+ self.assertRaises(exception.ProjectNotFound,
+ self.resource_api.get_project,
+ project_id)
+ # recreate project
+ self.resource_api.create_project(project_id, project)
+ self.resource_api.get_project(project_id)
+ # delete project
+ self.resource_api.delete_project(project_id)
+ # Verify ProjectNotFound is raised
+ self.assertRaises(exception.ProjectNotFound,
+ self.resource_api.get_project,
+ project_id)
+
+ def _assert_create_hierarchy_not_allowed(self):
+ domain = self._get_domain_fixture()
+
+ project1 = {'id': uuid.uuid4().hex,
+ 'name': uuid.uuid4().hex,
+ 'description': '',
+ 'domain_id': domain['id'],
+ 'enabled': True,
+ 'parent_id': None}
+ self.resource_api.create_project(project1['id'], project1)
+
+ # Creating project2 under project1. LDAP will not allow
+ # the creation of a project with parent_id being set
+ project2 = {'id': uuid.uuid4().hex,
+ 'name': uuid.uuid4().hex,
+ 'description': '',
+ 'domain_id': domain['id'],
+ 'enabled': True,
+ 'parent_id': project1['id']}
+
+ self.assertRaises(exception.InvalidParentProject,
+ self.resource_api.create_project,
+ project2['id'],
+ project2)
+
+ # Now, we'll create project 2 with no parent
+ project2['parent_id'] = None
+ self.resource_api.create_project(project2['id'], project2)
+
+ # Returning projects to be used across the tests
+ return [project1, project2]
+
+ def test_check_leaf_projects(self):
+ projects = self._assert_create_hierarchy_not_allowed()
+ for project in projects:
+ self.assertTrue(self.resource_api.is_leaf_project(project))
+
+ def test_list_projects_in_subtree(self):
+ projects = self._assert_create_hierarchy_not_allowed()
+ for project in projects:
+ subtree_list = self.resource_api.list_projects_in_subtree(
+ project)
+ self.assertEqual(0, len(subtree_list))
+
+ def test_list_project_parents(self):
+ projects = self._assert_create_hierarchy_not_allowed()
+ for project in projects:
+ parents_list = self.resource_api.list_project_parents(project)
+ self.assertEqual(0, len(parents_list))
+
+ def test_hierarchical_projects_crud(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_create_project_under_disabled_one(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_create_project_with_invalid_parent(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_create_leaf_project_with_invalid_domain(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_update_project_parent(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_enable_project_with_disabled_parent(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_disable_hierarchical_leaf_project(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_disable_hierarchical_not_leaf_project(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_delete_hierarchical_leaf_project(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_delete_hierarchical_not_leaf_project(self):
+ self._assert_create_hierarchy_not_allowed()
+
+ def test_check_hierarchy_depth(self):
+ projects = self._assert_create_hierarchy_not_allowed()
+ for project in projects:
+ depth = self._get_hierarchy_depth(project['id'])
+ self.assertEqual(1, depth)
+
+ def test_multi_role_grant_by_user_group_on_project_domain(self):
+ # This is a partial implementation of the standard test that
+ # is defined in test_backend.py. It omits both domain and
+ # group grants. since neither of these are yet supported by
+ # the ldap backend.
+
+ role_list = []
+ for _ in range(2):
+ role = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
+ self.role_api.create_role(role['id'], role)
+ role_list.append(role)
+
+ user1 = {'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id,
+ 'password': uuid.uuid4().hex,
+ 'enabled': True}
+ user1 = self.identity_api.create_user(user1)
+ project1 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id}
+ self.resource_api.create_project(project1['id'], project1)
+
+ self.assignment_api.add_role_to_user_and_project(
+ user_id=user1['id'],
+ tenant_id=project1['id'],
+ role_id=role_list[0]['id'])
+ self.assignment_api.add_role_to_user_and_project(
+ user_id=user1['id'],
+ tenant_id=project1['id'],
+ role_id=role_list[1]['id'])
+
+ # Although list_grants are not yet supported, we can test the
+ # alternate way of getting back lists of grants, where user
+ # and group roles are combined. Only directly assigned user
+ # roles are available, since group grants are not yet supported
+
+ combined_list = self.assignment_api.get_roles_for_user_and_project(
+ user1['id'],
+ project1['id'])
+ self.assertEqual(2, len(combined_list))
+ self.assertIn(role_list[0]['id'], combined_list)
+ self.assertIn(role_list[1]['id'], combined_list)
+
+ # Finally, although domain roles are not implemented, check we can
+ # issue the combined get roles call with benign results, since thus is
+ # used in token generation
+
+ combined_role_list = self.assignment_api.get_roles_for_user_and_domain(
+ user1['id'], CONF.identity.default_domain_id)
+ self.assertEqual(0, len(combined_role_list))
+
+ def test_list_projects_for_alternate_domain(self):
+ self.skipTest(
+ 'N/A: LDAP does not support multiple domains')
+
+ def test_get_default_domain_by_name(self):
+ domain = self._get_domain_fixture()
+
+ domain_ref = self.resource_api.get_domain_by_name(domain['name'])
+ self.assertEqual(domain_ref, domain)
+
+ def test_base_ldap_connection_deref_option(self):
+ def get_conn(deref_name):
+ self.config_fixture.config(group='ldap',
+ alias_dereferencing=deref_name)
+ base_ldap = common_ldap.BaseLdap(CONF)
+ return base_ldap.get_connection()
+
+ conn = get_conn('default')
+ self.assertEqual(ldap.get_option(ldap.OPT_DEREF),
+ conn.get_option(ldap.OPT_DEREF))
+
+ conn = get_conn('always')
+ self.assertEqual(ldap.DEREF_ALWAYS,
+ conn.get_option(ldap.OPT_DEREF))
+
+ conn = get_conn('finding')
+ self.assertEqual(ldap.DEREF_FINDING,
+ conn.get_option(ldap.OPT_DEREF))
+
+ conn = get_conn('never')
+ self.assertEqual(ldap.DEREF_NEVER,
+ conn.get_option(ldap.OPT_DEREF))
+
+ conn = get_conn('searching')
+ self.assertEqual(ldap.DEREF_SEARCHING,
+ conn.get_option(ldap.OPT_DEREF))
+
+ def test_list_users_no_dn(self):
+ users = self.identity_api.list_users()
+ self.assertEqual(len(default_fixtures.USERS), len(users))
+ user_ids = set(user['id'] for user in users)
+ expected_user_ids = set(getattr(self, 'user_%s' % user['id'])['id']
+ for user in default_fixtures.USERS)
+ for user_ref in users:
+ self.assertNotIn('dn', user_ref)
+ self.assertEqual(expected_user_ids, user_ids)
+
+ def test_list_groups_no_dn(self):
+ # Create some test groups.
+ domain = self._get_domain_fixture()
+ expected_group_ids = []
+ numgroups = 3
+ for _ in range(numgroups):
+ group = {'name': uuid.uuid4().hex, 'domain_id': domain['id']}
+ group = self.identity_api.create_group(group)
+ expected_group_ids.append(group['id'])
+ # Fetch the test groups and ensure that they don't contain a dn.
+ groups = self.identity_api.list_groups()
+ self.assertEqual(numgroups, len(groups))
+ group_ids = set(group['id'] for group in groups)
+ for group_ref in groups:
+ self.assertNotIn('dn', group_ref)
+ self.assertEqual(set(expected_group_ids), group_ids)
+
+ def test_list_groups_for_user_no_dn(self):
+ # Create a test user.
+ user = {'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id,
+ 'password': uuid.uuid4().hex, 'enabled': True}
+ user = self.identity_api.create_user(user)
+ # Create some test groups and add the test user as a member.
+ domain = self._get_domain_fixture()
+ expected_group_ids = []
+ numgroups = 3
+ for _ in range(numgroups):
+ group = {'name': uuid.uuid4().hex, 'domain_id': domain['id']}
+ group = self.identity_api.create_group(group)
+ expected_group_ids.append(group['id'])
+ self.identity_api.add_user_to_group(user['id'], group['id'])
+ # Fetch the groups for the test user
+ # and ensure they don't contain a dn.
+ groups = self.identity_api.list_groups_for_user(user['id'])
+ self.assertEqual(numgroups, len(groups))
+ group_ids = set(group['id'] for group in groups)
+ for group_ref in groups:
+ self.assertNotIn('dn', group_ref)
+ self.assertEqual(set(expected_group_ids), group_ids)
+
+ def test_user_id_attribute_in_create(self):
+ conf = self.get_config(CONF.identity.default_domain_id)
+ conf.ldap.user_id_attribute = 'mail'
+ self.reload_backends(CONF.identity.default_domain_id)
+
+ user = {'name': u'fäké1',
+ 'password': u'fäképass1',
+ 'domain_id': CONF.identity.default_domain_id}
+ user = self.identity_api.create_user(user)
+ user_ref = self.identity_api.get_user(user['id'])
+ # 'email' attribute should've created because it is also being used
+ # as user_id
+ self.assertEqual(user_ref['id'], user_ref['email'])
+
+ def test_user_id_attribute_map(self):
+ conf = self.get_config(CONF.identity.default_domain_id)
+ conf.ldap.user_id_attribute = 'mail'
+ self.reload_backends(CONF.identity.default_domain_id)
+
+ user_ref = self.identity_api.get_user(self.user_foo['email'])
+ # the user_id_attribute map should be honored, which means
+ # user_ref['id'] should contains the email attribute
+ self.assertEqual(self.user_foo['email'], user_ref['id'])
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_get_id_from_dn_for_multivalued_attribute_id(self, mock_ldap_get):
+ conf = self.get_config(CONF.identity.default_domain_id)
+ conf.ldap.user_id_attribute = 'mail'
+ self.reload_backends(CONF.identity.default_domain_id)
+
+ # make 'email' multivalued so we can test the error condition
+ email1 = uuid.uuid4().hex
+ email2 = uuid.uuid4().hex
+ mock_ldap_get.return_value = (
+ 'cn=nobodycares,dc=example,dc=com',
+ {
+ 'sn': [uuid.uuid4().hex],
+ 'mail': [email1, email2],
+ 'cn': 'nobodycares'
+ }
+ )
+
+ user_ref = self.identity_api.get_user(email1)
+ # make sure we get the ID from DN (old behavior) if the ID attribute
+ # has multiple values
+ self.assertEqual('nobodycares', user_ref['id'])
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_id_attribute_not_found(self, mock_ldap_get):
+ mock_ldap_get.return_value = (
+ 'cn=nobodycares,dc=example,dc=com',
+ {
+ 'sn': [uuid.uuid4().hex],
+ }
+ )
+
+ user_api = identity.backends.ldap.UserApi(CONF)
+ self.assertRaises(exception.NotFound,
+ user_api.get,
+ 'nobodycares')
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_user_id_not_in_dn(self, mock_ldap_get):
+ conf = self.get_config(CONF.identity.default_domain_id)
+ conf.ldap.user_id_attribute = 'uid'
+ conf.ldap.user_name_attribute = 'cn'
+ self.reload_backends(CONF.identity.default_domain_id)
+
+ mock_ldap_get.return_value = (
+ 'foo=bar,dc=example,dc=com',
+ {
+ 'sn': [uuid.uuid4().hex],
+ 'foo': ['bar'],
+ 'cn': ['junk'],
+ 'uid': ['crap']
+ }
+ )
+ user_ref = self.identity_api.get_user('crap')
+ self.assertEqual('crap', user_ref['id'])
+ self.assertEqual('junk', user_ref['name'])
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_user_name_in_dn(self, mock_ldap_get):
+ conf = self.get_config(CONF.identity.default_domain_id)
+ conf.ldap.user_id_attribute = 'sAMAccountName'
+ conf.ldap.user_name_attribute = 'cn'
+ self.reload_backends(CONF.identity.default_domain_id)
+
+ mock_ldap_get.return_value = (
+ 'cn=Foo Bar,dc=example,dc=com',
+ {
+ 'sn': [uuid.uuid4().hex],
+ 'cn': ['Foo Bar'],
+ 'SAMAccountName': ['crap']
+ }
+ )
+ user_ref = self.identity_api.get_user('crap')
+ self.assertEqual('crap', user_ref['id'])
+ self.assertEqual('Foo Bar', user_ref['name'])
+
+
+class LDAPIdentityEnabledEmulation(LDAPIdentity):
+ def setUp(self):
+ super(LDAPIdentityEnabledEmulation, self).setUp()
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+ for obj in [self.tenant_bar, self.tenant_baz, self.user_foo,
+ self.user_two, self.user_badguy]:
+ obj.setdefault('enabled', True)
+
+ def load_fixtures(self, fixtures):
+ # Override super impl since need to create group container.
+ create_group_container(self.identity_api)
+ super(LDAPIdentity, self).load_fixtures(fixtures)
+
+ def config_files(self):
+ config_files = super(LDAPIdentityEnabledEmulation, self).config_files()
+ config_files.append(tests.dirs.tests_conf('backend_ldap.conf'))
+ return config_files
+
+ def config_overrides(self):
+ super(LDAPIdentityEnabledEmulation, self).config_overrides()
+ self.config_fixture.config(group='ldap',
+ user_enabled_emulation=True,
+ project_enabled_emulation=True)
+
+ def test_project_crud(self):
+ # NOTE(topol): LDAPIdentityEnabledEmulation will create an
+ # enabled key in the project dictionary so this
+ # method override handles this side-effect
+ project = {
+ 'id': uuid.uuid4().hex,
+ 'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id,
+ 'description': uuid.uuid4().hex,
+ 'parent_id': None}
+
+ self.resource_api.create_project(project['id'], project)
+ project_ref = self.resource_api.get_project(project['id'])
+
+ # self.resource_api.create_project adds an enabled
+ # key with a value of True when LDAPIdentityEnabledEmulation
+ # is used so we now add this expected key to the project dictionary
+ project['enabled'] = True
+ self.assertDictEqual(project_ref, project)
+
+ project['description'] = uuid.uuid4().hex
+ self.resource_api.update_project(project['id'], project)
+ project_ref = self.resource_api.get_project(project['id'])
+ self.assertDictEqual(project_ref, project)
+
+ self.resource_api.delete_project(project['id'])
+ self.assertRaises(exception.ProjectNotFound,
+ self.resource_api.get_project,
+ project['id'])
+
+ def test_user_crud(self):
+ user_dict = {
+ 'domain_id': CONF.identity.default_domain_id,
+ 'name': uuid.uuid4().hex,
+ 'password': uuid.uuid4().hex}
+ user = self.identity_api.create_user(user_dict)
+ user_dict['enabled'] = True
+ user_ref = self.identity_api.get_user(user['id'])
+ del user_dict['password']
+ user_ref_dict = {x: user_ref[x] for x in user_ref}
+ self.assertDictContainsSubset(user_dict, user_ref_dict)
+
+ user_dict['password'] = uuid.uuid4().hex
+ self.identity_api.update_user(user['id'], user)
+ user_ref = self.identity_api.get_user(user['id'])
+ del user_dict['password']
+ user_ref_dict = {x: user_ref[x] for x in user_ref}
+ self.assertDictContainsSubset(user_dict, user_ref_dict)
+
+ self.identity_api.delete_user(user['id'])
+ self.assertRaises(exception.UserNotFound,
+ self.identity_api.get_user,
+ user['id'])
+
+ def test_user_auth_emulated(self):
+ self.config_fixture.config(group='ldap',
+ user_enabled_emulation_dn='cn=test,dc=test')
+ self.reload_backends(CONF.identity.default_domain_id)
+ self.identity_api.authenticate(
+ context={},
+ user_id=self.user_foo['id'],
+ password=self.user_foo['password'])
+
+ def test_user_enable_attribute_mask(self):
+ self.skipTest(
+ "Enabled emulation conflicts with enabled mask")
+
+ def test_user_enabled_invert(self):
+ self.config_fixture.config(group='ldap', user_enabled_invert=True,
+ user_enabled_default=False)
+ self.clear_database()
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+
+ user1 = {'name': u'fäké1', 'enabled': True,
+ 'domain_id': CONF.identity.default_domain_id}
+
+ user2 = {'name': u'fäké2', 'enabled': False,
+ 'domain_id': CONF.identity.default_domain_id}
+
+ user3 = {'name': u'fäké3',
+ 'domain_id': CONF.identity.default_domain_id}
+
+ # Ensure that the enabled LDAP attribute is not set for a
+ # newly created enabled user.
+ user_ref = self.identity_api.create_user(user1)
+ self.assertIs(True, user_ref['enabled'])
+ self.assertIsNone(self.get_user_enabled_vals(user_ref))
+ user_ref = self.identity_api.get_user(user_ref['id'])
+ self.assertIs(True, user_ref['enabled'])
+
+ # Ensure that an enabled LDAP attribute is not set for a disabled user.
+ user1['enabled'] = False
+ user_ref = self.identity_api.update_user(user_ref['id'], user1)
+ self.assertIs(False, user_ref['enabled'])
+ self.assertIsNone(self.get_user_enabled_vals(user_ref))
+
+ # Enable the user and ensure that the LDAP enabled
+ # attribute is not set.
+ user1['enabled'] = True
+ user_ref = self.identity_api.update_user(user_ref['id'], user1)
+ self.assertIs(True, user_ref['enabled'])
+ self.assertIsNone(self.get_user_enabled_vals(user_ref))
+
+ # Ensure that the LDAP enabled attribute is not set for a
+ # newly created disabled user.
+ user_ref = self.identity_api.create_user(user2)
+ self.assertIs(False, user_ref['enabled'])
+ self.assertIsNone(self.get_user_enabled_vals(user_ref))
+ user_ref = self.identity_api.get_user(user_ref['id'])
+ self.assertIs(False, user_ref['enabled'])
+
+ # Ensure that the LDAP enabled attribute is not set for a newly created
+ # user when the user_enabled_default setting is used.
+ user_ref = self.identity_api.create_user(user3)
+ self.assertIs(True, user_ref['enabled'])
+ self.assertIsNone(self.get_user_enabled_vals(user_ref))
+ user_ref = self.identity_api.get_user(user_ref['id'])
+ self.assertIs(True, user_ref['enabled'])
+
+ def test_user_enabled_invert_no_enabled_value(self):
+ self.skipTest(
+ "N/A: Covered by test_user_enabled_invert")
+
+ def test_user_enabled_invert_default_str_value(self):
+ self.skipTest(
+ "N/A: Covered by test_user_enabled_invert")
+
+ @mock.patch.object(common_ldap_core.BaseLdap, '_ldap_get')
+ def test_user_enabled_attribute_handles_utf8(self, mock_ldap_get):
+ # Since user_enabled_emulation is enabled in this test, this test will
+ # fail since it's using user_enabled_invert.
+ self.config_fixture.config(group='ldap', user_enabled_invert=True,
+ user_enabled_attribute='passwordisexpired')
+ mock_ldap_get.return_value = (
+ u'uid=123456789,c=us,ou=our_ldap,o=acme.com',
+ {
+ 'uid': [123456789],
+ 'mail': [u'shaun@acme.com'],
+ 'passwordisexpired': [u'false'],
+ 'cn': [u'uid=123456789,c=us,ou=our_ldap,o=acme.com']
+ }
+ )
+
+ user_api = identity.backends.ldap.UserApi(CONF)
+ user_ref = user_api.get('123456789')
+ self.assertIs(False, user_ref['enabled'])
+
+
+class LdapIdentitySqlAssignment(BaseLDAPIdentity, tests.SQLDriverOverrides,
+ tests.TestCase):
+
+ def config_files(self):
+ config_files = super(LdapIdentitySqlAssignment, self).config_files()
+ config_files.append(tests.dirs.tests_conf('backend_ldap_sql.conf'))
+ return config_files
+
+ def setUp(self):
+ self.useFixture(database.Database())
+ super(LdapIdentitySqlAssignment, self).setUp()
+ self.clear_database()
+ self.load_backends()
+ cache.configure_cache_region(cache.REGION)
+ self.engine = sql.get_engine()
+ self.addCleanup(sql.cleanup)
+
+ sql.ModelBase.metadata.create_all(bind=self.engine)
+ self.addCleanup(sql.ModelBase.metadata.drop_all, bind=self.engine)
+
+ self.load_fixtures(default_fixtures)
+ # defaulted by the data load
+ self.user_foo['enabled'] = True
+
+ def config_overrides(self):
+ super(LdapIdentitySqlAssignment, self).config_overrides()
+ self.config_fixture.config(
+ group='identity',
+ driver='keystone.identity.backends.ldap.Identity')
+ self.config_fixture.config(
+ group='resource',
+ driver='keystone.resource.backends.sql.Resource')
+ self.config_fixture.config(
+ group='assignment',
+ driver='keystone.assignment.backends.sql.Assignment')
+
+ def test_domain_crud(self):
+ pass
+
+ def test_list_domains(self):
+ domains = self.resource_api.list_domains()
+ self.assertEqual([resource.calc_default_domain()], domains)
+
+ def test_list_domains_non_default_domain_id(self):
+ # If change the default_domain_id, the ID of the default domain
+ # returned by list_domains doesn't change because the SQL identity
+ # backend reads it from the database, which doesn't get updated by
+ # config change.
+
+ orig_default_domain_id = CONF.identity.default_domain_id
+
+ new_domain_id = uuid.uuid4().hex
+ self.config_fixture.config(group='identity',
+ default_domain_id=new_domain_id)
+
+ domains = self.resource_api.list_domains()
+
+ self.assertEqual(orig_default_domain_id, domains[0]['id'])
+
+ def test_create_domain(self):
+ domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
+ 'enabled': True}
+ self.assertRaises(exception.Forbidden,
+ self.resource_api.create_domain,
+ domain['id'],
+ domain)
+
+ def test_get_and_remove_role_grant_by_group_and_domain(self):
+ # TODO(henry-nash): We should really rewrite the tests in test_backend
+ # to be more flexible as to where the domains are sourced from, so
+ # that we would not need to override such tests here. This is raised
+ # as bug 1373865.
+ new_domain = self._get_domain_fixture()
+ new_group = {'domain_id': new_domain['id'], 'name': uuid.uuid4().hex}
+ new_group = self.identity_api.create_group(new_group)
+ new_user = {'name': 'new_user', 'password': uuid.uuid4().hex,
+ 'enabled': True, 'domain_id': new_domain['id']}
+ new_user = self.identity_api.create_user(new_user)
+ self.identity_api.add_user_to_group(new_user['id'],
+ new_group['id'])
+
+ roles_ref = self.assignment_api.list_grants(
+ group_id=new_group['id'],
+ domain_id=new_domain['id'])
+ self.assertEqual(0, len(roles_ref))
+
+ self.assignment_api.create_grant(group_id=new_group['id'],
+ domain_id=new_domain['id'],
+ role_id='member')
+
+ roles_ref = self.assignment_api.list_grants(
+ group_id=new_group['id'],
+ domain_id=new_domain['id'])
+ self.assertDictEqual(roles_ref[0], self.role_member)
+
+ self.assignment_api.delete_grant(group_id=new_group['id'],
+ domain_id=new_domain['id'],
+ role_id='member')
+ roles_ref = self.assignment_api.list_grants(
+ group_id=new_group['id'],
+ domain_id=new_domain['id'])
+ self.assertEqual(0, len(roles_ref))
+ self.assertRaises(exception.NotFound,
+ self.assignment_api.delete_grant,
+ group_id=new_group['id'],
+ domain_id=new_domain['id'],
+ role_id='member')
+
+ def test_project_enabled_ignored_disable_error(self):
+ # Override
+ self.skipTest("Doesn't apply since LDAP configuration is ignored for "
+ "SQL assignment backend.")
+
+
+class LdapIdentitySqlAssignmentWithMapping(LdapIdentitySqlAssignment):
+ """Class to test mapping of default LDAP backend.
+
+ The default configuration is not to enable mapping when using a single
+ backend LDAP driver. However, a cloud provider might want to enable
+ the mapping, hence hiding the LDAP IDs from any clients of keystone.
+ Setting backward_compatible_ids to False will enable this mapping.
+
+ """
+ def config_overrides(self):
+ super(LdapIdentitySqlAssignmentWithMapping, self).config_overrides()
+ self.config_fixture.config(group='identity_mapping',
+ backward_compatible_ids=False)
+
+ def test_dynamic_mapping_build(self):
+ """Test to ensure entities not create via controller are mapped.
+
+ Many LDAP backends will, essentially, by Read Only. In these cases
+ the mapping is not built by creating objects, rather from enumerating
+ the entries. We test this here my manually deleting the mapping and
+ then trying to re-read the entries.
+
+ """
+ initial_mappings = len(mapping_sql.list_id_mappings())
+ user1 = {'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id,
+ 'password': uuid.uuid4().hex, 'enabled': True}
+ user1 = self.identity_api.create_user(user1)
+ user2 = {'name': uuid.uuid4().hex,
+ 'domain_id': CONF.identity.default_domain_id,
+ 'password': uuid.uuid4().hex, 'enabled': True}
+ user2 = self.identity_api.create_user(user2)
+ mappings = mapping_sql.list_id_mappings()
+ self.assertEqual(initial_mappings + 2, len(mappings))
+
+ # Now delete the mappings for the two users above
+ self.id_mapping_api.purge_mappings({'public_id': user1['id']})
+ self.id_mapping_api.purge_mappings({'public_id': user2['id']})
+
+ # We should no longer be able to get these users via their old IDs
+ self.assertRaises(exception.UserNotFound,
+ self.identity_api.get_user,
+ user1['id'])
+ self.assertRaises(exception.UserNotFound,
+ self.identity_api.get_user,
+ user2['id'])
+
+ # Now enumerate all users...this should re-build the mapping, and
+ # we should be able to find the users via their original public IDs.
+ self.identity_api.list_users()
+ self.identity_api.get_user(user1['id'])
+ self.identity_api.get_user(user2['id'])
+
+ def test_get_roles_for_user_and_project_user_group_same_id(self):
+ self.skipTest('N/A: We never generate the same ID for a user and '
+ 'group in our mapping table')
+
+
+class BaseMultiLDAPandSQLIdentity(object):
+ """Mixin class with support methods for domain-specific config testing."""
+
+ def create_user(self, domain_id):
+ user = {'name': uuid.uuid4().hex,
+ 'domain_id': domain_id,
+ 'password': uuid.uuid4().hex,
+ 'enabled': True}
+ user_ref = self.identity_api.create_user(user)
+ # Put the password back in, since this is used later by tests to
+ # authenticate.
+ user_ref['password'] = user['password']
+ return user_ref
+
+ def create_users_across_domains(self):
+ """Create a set of users, each with a role on their own domain."""
+
+ # We also will check that the right number of id mappings get created
+ initial_mappings = len(mapping_sql.list_id_mappings())
+
+ self.users['user0'] = self.create_user(
+ self.domains['domain_default']['id'])
+ self.assignment_api.create_grant(
+ user_id=self.users['user0']['id'],
+ domain_id=self.domains['domain_default']['id'],
+ role_id=self.role_member['id'])
+ for x in range(1, self.domain_count):
+ self.users['user%s' % x] = self.create_user(
+ self.domains['domain%s' % x]['id'])
+ self.assignment_api.create_grant(
+ user_id=self.users['user%s' % x]['id'],
+ domain_id=self.domains['domain%s' % x]['id'],
+ role_id=self.role_member['id'])
+
+ # So how many new id mappings should have been created? One for each
+ # user created in a domain that is using the non default driver..
+ self.assertEqual(initial_mappings + self.domain_specific_count,
+ len(mapping_sql.list_id_mappings()))
+
+ def check_user(self, user, domain_id, expected_status):
+ """Check user is in correct backend.
+
+ As part of the tests, we want to force ourselves to manually
+ select the driver for a given domain, to make sure the entity
+ ended up in the correct backend.
+
+ """
+ driver = self.identity_api._select_identity_driver(domain_id)
+ unused, unused, entity_id = (
+ self.identity_api._get_domain_driver_and_entity_id(
+ user['id']))
+
+ if expected_status == 200:
+ ref = driver.get_user(entity_id)
+ ref = self.identity_api._set_domain_id_and_mapping(
+ ref, domain_id, driver, map.EntityType.USER)
+ user = user.copy()
+ del user['password']
+ self.assertDictEqual(ref, user)
+ else:
+ # TODO(henry-nash): Use AssertRaises here, although
+ # there appears to be an issue with using driver.get_user
+ # inside that construct
+ try:
+ driver.get_user(entity_id)
+ except expected_status:
+ pass
+
+ def setup_initial_domains(self):
+
+ def create_domain(domain):
+ try:
+ ref = self.resource_api.create_domain(
+ domain['id'], domain)
+ except exception.Conflict:
+ ref = (
+ self.resource_api.get_domain_by_name(domain['name']))
+ return ref
+
+ self.domains = {}
+ for x in range(1, self.domain_count):
+ domain = 'domain%s' % x
+ self.domains[domain] = create_domain(
+ {'id': uuid.uuid4().hex, 'name': domain})
+ self.domains['domain_default'] = create_domain(
+ resource.calc_default_domain())
+
+ def test_authenticate_to_each_domain(self):
+ """Test that a user in each domain can authenticate."""
+ for user_num in range(self.domain_count):
+ user = 'user%s' % user_num
+ self.identity_api.authenticate(
+ context={},
+ user_id=self.users[user]['id'],
+ password=self.users[user]['password'])
+
+
+class MultiLDAPandSQLIdentity(BaseLDAPIdentity, tests.SQLDriverOverrides,
+ tests.TestCase, BaseMultiLDAPandSQLIdentity):
+ """Class to test common SQL plus individual LDAP backends.
+
+ We define a set of domains and domain-specific backends:
+
+ - A separate LDAP backend for the default domain
+ - A separate LDAP backend for domain1
+ - domain2 shares the same LDAP as domain1, but uses a different
+ tree attach point
+ - An SQL backend for all other domains (which will include domain3
+ and domain4)
+
+ Normally one would expect that the default domain would be handled as
+ part of the "other domains" - however the above provides better
+ test coverage since most of the existing backend tests use the default
+ domain.
+
+ """
+ def setUp(self):
+ self.useFixture(database.Database())
+ super(MultiLDAPandSQLIdentity, self).setUp()
+
+ self.load_backends()
+
+ self.engine = sql.get_engine()
+ self.addCleanup(sql.cleanup)
+
+ sql.ModelBase.metadata.create_all(bind=self.engine)
+ self.addCleanup(sql.ModelBase.metadata.drop_all, bind=self.engine)
+
+ self.domain_count = 5
+ self.domain_specific_count = 3
+ self.setup_initial_domains()
+ self._setup_initial_users()
+
+ # All initial test data setup complete, time to switch on support
+ # for separate backends per domain.
+ self.enable_multi_domain()
+
+ self.clear_database()
+ self.load_fixtures(default_fixtures)
+ self.create_users_across_domains()
+
+ def config_overrides(self):
+ super(MultiLDAPandSQLIdentity, self).config_overrides()
+ # Make sure identity and assignment are actually SQL drivers,
+ # BaseLDAPIdentity sets these options to use LDAP.
+ self.config_fixture.config(
+ group='identity',
+ driver='keystone.identity.backends.sql.Identity')
+ self.config_fixture.config(
+ group='resource',
+ driver='keystone.resource.backends.sql.Resource')
+ self.config_fixture.config(
+ group='assignment',
+ driver='keystone.assignment.backends.sql.Assignment')
+
+ def _setup_initial_users(self):
+ # Create some identity entities BEFORE we switch to multi-backend, so
+ # we can test that these are still accessible
+ self.users = {}
+ self.users['userA'] = self.create_user(
+ self.domains['domain_default']['id'])
+ self.users['userB'] = self.create_user(
+ self.domains['domain1']['id'])
+ self.users['userC'] = self.create_user(
+ self.domains['domain3']['id'])
+
+ def enable_multi_domain(self):
+ """Enable the chosen form of multi domain configuration support.
+
+ This method enables the file-based configuration support. Child classes
+ that wish to use the database domain configuration support should
+ override this method and set the appropriate config_fixture option.
+
+ """
+ self.config_fixture.config(
+ group='identity', domain_specific_drivers_enabled=True,
+ domain_config_dir=tests.TESTCONF + '/domain_configs_multi_ldap')
+ self.config_fixture.config(group='identity_mapping',
+ backward_compatible_ids=False)
+
+ def reload_backends(self, domain_id):
+ # Just reload the driver for this domain - which will pickup
+ # any updated cfg
+ self.identity_api.domain_configs.reload_domain_driver(domain_id)
+
+ def get_config(self, domain_id):
+ # Get the config for this domain, will return CONF
+ # if no specific config defined for this domain
+ return self.identity_api.domain_configs.get_domain_conf(domain_id)
+
+ def test_list_domains(self):
+ self.skipTest(
+ 'N/A: Not relevant for multi ldap testing')
+
+ def test_list_domains_non_default_domain_id(self):
+ self.skipTest(
+ 'N/A: Not relevant for multi ldap testing')
+
+ def test_list_users(self):
+ # Override the standard list users, since we have added an extra user
+ # to the default domain, so the number of expected users is one more
+ # than in the standard test.
+ users = self.identity_api.list_users(
+ domain_scope=self._set_domain_scope(
+ CONF.identity.default_domain_id))
+ self.assertEqual(len(default_fixtures.USERS) + 1, len(users))
+ user_ids = set(user['id'] for user in users)
+ expected_user_ids = set(getattr(self, 'user_%s' % user['id'])['id']
+ for user in default_fixtures.USERS)
+ expected_user_ids.add(self.users['user0']['id'])
+ for user_ref in users:
+ self.assertNotIn('password', user_ref)
+ self.assertEqual(expected_user_ids, user_ids)
+
+ def test_domain_segregation(self):
+ """Test that separate configs have segregated the domain.
+
+ Test Plan:
+
+ - Users were created in each domain as part of setup, now make sure
+ you can only find a given user in its relevant domain/backend
+ - Make sure that for a backend that supports multiple domains
+ you can get the users via any of its domains
+
+ """
+ # Check that I can read a user with the appropriate domain-selected
+ # driver, but won't find it via any other domain driver
+
+ check_user = self.check_user
+ check_user(self.users['user0'],
+ self.domains['domain_default']['id'], 200)
+ for domain in [self.domains['domain1']['id'],
+ self.domains['domain2']['id'],
+ self.domains['domain3']['id'],
+ self.domains['domain4']['id']]:
+ check_user(self.users['user0'], domain, exception.UserNotFound)
+
+ check_user(self.users['user1'], self.domains['domain1']['id'], 200)
+ for domain in [self.domains['domain_default']['id'],
+ self.domains['domain2']['id'],
+ self.domains['domain3']['id'],
+ self.domains['domain4']['id']]:
+ check_user(self.users['user1'], domain, exception.UserNotFound)
+
+ check_user(self.users['user2'], self.domains['domain2']['id'], 200)
+ for domain in [self.domains['domain_default']['id'],
+ self.domains['domain1']['id'],
+ self.domains['domain3']['id'],
+ self.domains['domain4']['id']]:
+ check_user(self.users['user2'], domain, exception.UserNotFound)
+
+ # domain3 and domain4 share the same backend, so you should be
+ # able to see user3 and user4 from either.
+
+ check_user(self.users['user3'], self.domains['domain3']['id'], 200)
+ check_user(self.users['user3'], self.domains['domain4']['id'], 200)
+ check_user(self.users['user4'], self.domains['domain3']['id'], 200)
+ check_user(self.users['user4'], self.domains['domain4']['id'], 200)
+
+ for domain in [self.domains['domain_default']['id'],
+ self.domains['domain1']['id'],
+ self.domains['domain2']['id']]:
+ check_user(self.users['user3'], domain, exception.UserNotFound)
+ check_user(self.users['user4'], domain, exception.UserNotFound)
+
+ # Finally, going through the regular manager layer, make sure we
+ # only see the right number of users in each of the non-default
+ # domains. One might have expected two users in domain1 (since we
+ # created one before we switched to multi-backend), however since
+ # that domain changed backends in the switch we don't find it anymore.
+ # This is as designed - we don't support moving domains between
+ # backends.
+ #
+ # The listing of the default domain is already handled in the
+ # test_lists_users() method.
+ for domain in [self.domains['domain1']['id'],
+ self.domains['domain2']['id'],
+ self.domains['domain4']['id']]:
+ self.assertThat(
+ self.identity_api.list_users(domain_scope=domain),
+ matchers.HasLength(1))
+
+ # domain3 had a user created before we switched on
+ # multiple backends, plus one created afterwards - and its
+ # backend has not changed - so we should find two.
+ self.assertThat(
+ self.identity_api.list_users(
+ domain_scope=self.domains['domain3']['id']),
+ matchers.HasLength(2))
+
+ def test_existing_uuids_work(self):
+ """Test that 'uni-domain' created IDs still work.
+
+ Throwing the switch to domain-specific backends should not cause
+ existing identities to be inaccessible via ID.
+
+ """
+ self.identity_api.get_user(self.users['userA']['id'])
+ self.identity_api.get_user(self.users['userB']['id'])
+ self.identity_api.get_user(self.users['userC']['id'])
+
+ def test_scanning_of_config_dir(self):
+ """Test the Manager class scans the config directory.
+
+ The setup for the main tests above load the domain configs directly
+ so that the test overrides can be included. This test just makes sure
+ that the standard config directory scanning does pick up the relevant
+ domain config files.
+
+ """
+ # Confirm that config has drivers_enabled as True, which we will
+ # check has been set to False later in this test
+ self.assertTrue(CONF.identity.domain_specific_drivers_enabled)
+ self.load_backends()
+ # Execute any command to trigger the lazy loading of domain configs
+ self.identity_api.list_users(
+ domain_scope=self.domains['domain1']['id'])
+ # ...and now check the domain configs have been set up
+ self.assertIn('default', self.identity_api.domain_configs)
+ self.assertIn(self.domains['domain1']['id'],
+ self.identity_api.domain_configs)
+ self.assertIn(self.domains['domain2']['id'],
+ self.identity_api.domain_configs)
+ self.assertNotIn(self.domains['domain3']['id'],
+ self.identity_api.domain_configs)
+ self.assertNotIn(self.domains['domain4']['id'],
+ self.identity_api.domain_configs)
+
+ # Finally check that a domain specific config contains items from both
+ # the primary config and the domain specific config
+ conf = self.identity_api.domain_configs.get_domain_conf(
+ self.domains['domain1']['id'])
+ # This should now be false, as is the default, since this is not
+ # set in the standard primary config file
+ self.assertFalse(conf.identity.domain_specific_drivers_enabled)
+ # ..and make sure a domain-specific options is also set
+ self.assertEqual('fake://memory1', conf.ldap.url)
+
+ def test_delete_domain_with_user_added(self):
+ domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
+ 'enabled': True}
+ project = {'id': uuid.uuid4().hex,
+ 'name': uuid.uuid4().hex,
+ 'domain_id': domain['id'],
+ 'description': uuid.uuid4().hex,
+ 'parent_id': None,
+ 'enabled': True}
+ self.resource_api.create_domain(domain['id'], domain)
+ self.resource_api.create_project(project['id'], project)
+ project_ref = self.resource_api.get_project(project['id'])
+ self.assertDictEqual(project_ref, project)
+
+ self.assignment_api.create_grant(user_id=self.user_foo['id'],
+ project_id=project['id'],
+ role_id=self.role_member['id'])
+ self.assignment_api.delete_grant(user_id=self.user_foo['id'],
+ project_id=project['id'],
+ role_id=self.role_member['id'])
+ domain['enabled'] = False
+ self.resource_api.update_domain(domain['id'], domain)
+ self.resource_api.delete_domain(domain['id'])
+ self.assertRaises(exception.DomainNotFound,
+ self.resource_api.get_domain,
+ domain['id'])
+
+ def test_user_enabled_ignored_disable_error(self):
+ # Override.
+ self.skipTest("Doesn't apply since LDAP config has no affect on the "
+ "SQL identity backend.")
+
+ def test_group_enabled_ignored_disable_error(self):
+ # Override.
+ self.skipTest("Doesn't apply since LDAP config has no affect on the "
+ "SQL identity backend.")
+
+ def test_project_enabled_ignored_disable_error(self):
+ # Override
+ self.skipTest("Doesn't apply since LDAP configuration is ignored for "
+ "SQL assignment backend.")
+
+
+class MultiLDAPandSQLIdentityDomainConfigsInSQL(MultiLDAPandSQLIdentity):
+ """Class to test the use of domain configs stored in the database.
+
+ Repeat the same tests as MultiLDAPandSQLIdentity, but instead of using the
+ domain specific config files, store the domain specific values in the
+ database.
+
+ """
+ def enable_multi_domain(self):
+ # The values below are the same as in the domain_configs_multi_ldap
+ # cdirectory of test config_files.
+ default_config = {
+ 'ldap': {'url': 'fake://memory',
+ 'user': 'cn=Admin',
+ 'password': 'password',
+ 'suffix': 'cn=example,cn=com'},
+ 'identity': {'driver': 'keystone.identity.backends.ldap.Identity'}
+ }
+ domain1_config = {
+ 'ldap': {'url': 'fake://memory1',
+ 'user': 'cn=Admin',
+ 'password': 'password',
+ 'suffix': 'cn=example,cn=com'},
+ 'identity': {'driver': 'keystone.identity.backends.ldap.Identity'}
+ }
+ domain2_config = {
+ 'ldap': {'url': 'fake://memory',
+ 'user': 'cn=Admin',
+ 'password': 'password',
+ 'suffix': 'cn=myroot,cn=com',
+ 'group_tree_dn': 'ou=UserGroups,dc=myroot,dc=org',
+ 'user_tree_dn': 'ou=Users,dc=myroot,dc=org'},
+ 'identity': {'driver': 'keystone.identity.backends.ldap.Identity'}
+ }
+
+ self.domain_config_api.create_config(CONF.identity.default_domain_id,
+ default_config)
+ self.domain_config_api.create_config(self.domains['domain1']['id'],
+ domain1_config)
+ self.domain_config_api.create_config(self.domains['domain2']['id'],
+ domain2_config)
+
+ self.config_fixture.config(
+ group='identity', domain_specific_drivers_enabled=True,
+ domain_configurations_from_database=True)
+ self.config_fixture.config(group='identity_mapping',
+ backward_compatible_ids=False)
+
+ def test_domain_config_has_no_impact_if_database_support_disabled(self):
+ """Ensure database domain configs have no effect if disabled.
+
+ Set reading from database configs to false, restart the backends
+ and then try and set and use database configs.
+
+ """
+ self.config_fixture.config(
+ group='identity', domain_configurations_from_database=False)
+ self.load_backends()
+ new_config = {'ldap': {'url': uuid.uuid4().hex}}
+ self.domain_config_api.create_config(
+ CONF.identity.default_domain_id, new_config)
+ # Trigger the identity backend to initialise any domain specific
+ # configurations
+ self.identity_api.list_users()
+ # Check that the new config has not been passed to the driver for
+ # the default domain.
+ default_config = (
+ self.identity_api.domain_configs.get_domain_conf(
+ CONF.identity.default_domain_id))
+ self.assertEqual(CONF.ldap.url, default_config.ldap.url)
+
+
+class DomainSpecificLDAPandSQLIdentity(
+ BaseLDAPIdentity, tests.SQLDriverOverrides, tests.TestCase,
+ BaseMultiLDAPandSQLIdentity):
+ """Class to test when all domains use specific configs, including SQL.
+
+ We define a set of domains and domain-specific backends:
+
+ - A separate LDAP backend for the default domain
+ - A separate SQL backend for domain1
+
+ Although the default driver still exists, we don't use it.
+
+ """
+ def setUp(self):
+ self.useFixture(database.Database())
+ super(DomainSpecificLDAPandSQLIdentity, self).setUp()
+ self.initial_setup()
+
+ def initial_setup(self):
+ # We aren't setting up any initial data ahead of switching to
+ # domain-specific operation, so make the switch straight away.
+ self.config_fixture.config(
+ group='identity', domain_specific_drivers_enabled=True,
+ domain_config_dir=(
+ tests.TESTCONF + '/domain_configs_one_sql_one_ldap'))
+ self.config_fixture.config(group='identity_mapping',
+ backward_compatible_ids=False)
+
+ self.load_backends()
+
+ self.engine = sql.get_engine()
+ self.addCleanup(sql.cleanup)
+
+ sql.ModelBase.metadata.create_all(bind=self.engine)
+ self.addCleanup(sql.ModelBase.metadata.drop_all, bind=self.engine)
+
+ self.domain_count = 2
+ self.domain_specific_count = 2
+ self.setup_initial_domains()
+ self.users = {}
+
+ self.clear_database()
+ self.load_fixtures(default_fixtures)
+ self.create_users_across_domains()
+
+ def config_overrides(self):
+ super(DomainSpecificLDAPandSQLIdentity, self).config_overrides()
+ # Make sure resource & assignment are actually SQL drivers,
+ # BaseLDAPIdentity causes this option to use LDAP.
+ self.config_fixture.config(
+ group='resource',
+ driver='keystone.resource.backends.sql.Resource')
+ self.config_fixture.config(
+ group='assignment',
+ driver='keystone.assignment.backends.sql.Assignment')
+
+ def reload_backends(self, domain_id):
+ # Just reload the driver for this domain - which will pickup
+ # any updated cfg
+ self.identity_api.domain_configs.reload_domain_driver(domain_id)
+
+ def get_config(self, domain_id):
+ # Get the config for this domain, will return CONF
+ # if no specific config defined for this domain
+ return self.identity_api.domain_configs.get_domain_conf(domain_id)
+
+ def test_list_domains(self):
+ self.skipTest(
+ 'N/A: Not relevant for multi ldap testing')
+
+ def test_list_domains_non_default_domain_id(self):
+ self.skipTest(
+ 'N/A: Not relevant for multi ldap testing')
+
+ def test_domain_crud(self):
+ self.skipTest(
+ 'N/A: Not relevant for multi ldap testing')
+
+ def test_list_users(self):
+ # Override the standard list users, since we have added an extra user
+ # to the default domain, so the number of expected users is one more
+ # than in the standard test.
+ users = self.identity_api.list_users(
+ domain_scope=self._set_domain_scope(
+ CONF.identity.default_domain_id))
+ self.assertEqual(len(default_fixtures.USERS) + 1, len(users))
+ user_ids = set(user['id'] for user in users)
+ expected_user_ids = set(getattr(self, 'user_%s' % user['id'])['id']
+ for user in default_fixtures.USERS)
+ expected_user_ids.add(self.users['user0']['id'])
+ for user_ref in users:
+ self.assertNotIn('password', user_ref)
+ self.assertEqual(expected_user_ids, user_ids)
+
+ def test_domain_segregation(self):
+ """Test that separate configs have segregated the domain.
+
+ Test Plan:
+
+ - Users were created in each domain as part of setup, now make sure
+ you can only find a given user in its relevant domain/backend
+ - Make sure that for a backend that supports multiple domains
+ you can get the users via any of its domains
+
+ """
+ # Check that I can read a user with the appropriate domain-selected
+ # driver, but won't find it via any other domain driver
+
+ self.check_user(self.users['user0'],
+ self.domains['domain_default']['id'], 200)
+ self.check_user(self.users['user0'],
+ self.domains['domain1']['id'], exception.UserNotFound)
+
+ self.check_user(self.users['user1'],
+ self.domains['domain1']['id'], 200)
+ self.check_user(self.users['user1'],
+ self.domains['domain_default']['id'],
+ exception.UserNotFound)
+
+ # Finally, going through the regular manager layer, make sure we
+ # only see the right number of users in the non-default domain.
+
+ self.assertThat(
+ self.identity_api.list_users(
+ domain_scope=self.domains['domain1']['id']),
+ matchers.HasLength(1))
+
+ def test_add_role_grant_to_user_and_project_404(self):
+ self.skipTest('Blocked by bug 1101287')
+
+ def test_get_role_grants_for_user_and_project_404(self):
+ self.skipTest('Blocked by bug 1101287')
+
+ def test_list_projects_for_user_with_grants(self):
+ self.skipTest('Blocked by bug 1221805')
+
+ def test_get_roles_for_user_and_project_user_group_same_id(self):
+ self.skipTest('N/A: We never generate the same ID for a user and '
+ 'group in our mapping table')
+
+ def test_user_id_comma(self):
+ self.skipTest('Only valid if it is guaranteed to be talking to '
+ 'the fakeldap backend')
+
+ def test_user_id_comma_grants(self):
+ self.skipTest('Only valid if it is guaranteed to be talking to '
+ 'the fakeldap backend')
+
+ def test_user_enabled_ignored_disable_error(self):
+ # Override.
+ self.skipTest("Doesn't apply since LDAP config has no affect on the "
+ "SQL identity backend.")
+
+ def test_group_enabled_ignored_disable_error(self):
+ # Override.
+ self.skipTest("Doesn't apply since LDAP config has no affect on the "
+ "SQL identity backend.")
+
+ def test_project_enabled_ignored_disable_error(self):
+ # Override
+ self.skipTest("Doesn't apply since LDAP configuration is ignored for "
+ "SQL assignment backend.")
+
+
+class DomainSpecificSQLIdentity(DomainSpecificLDAPandSQLIdentity):
+ """Class to test simplest use of domain-specific SQL driver.
+
+ The simplest use of an SQL domain-specific backend is when it is used to
+ augment the standard case when LDAP is the default driver defined in the
+ main config file. This would allow, for example, service users to be
+ stored in SQL while LDAP handles the rest. Hence we define:
+
+ - The default driver uses the LDAP backend for the default domain
+ - A separate SQL backend for domain1
+
+ """
+ def initial_setup(self):
+ # We aren't setting up any initial data ahead of switching to
+ # domain-specific operation, so make the switch straight away.
+ self.config_fixture.config(
+ group='identity', domain_specific_drivers_enabled=True,
+ domain_config_dir=(
+ tests.TESTCONF + '/domain_configs_default_ldap_one_sql'))
+ # Part of the testing counts how many new mappings get created as
+ # we create users, so ensure we are NOT using mapping for the default
+ # LDAP domain so this doesn't confuse the calculation.
+ self.config_fixture.config(group='identity_mapping',
+ backward_compatible_ids=True)
+
+ self.load_backends()
+
+ self.engine = sql.get_engine()
+ self.addCleanup(sql.cleanup)
+
+ sql.ModelBase.metadata.create_all(bind=self.engine)
+ self.addCleanup(sql.ModelBase.metadata.drop_all, bind=self.engine)
+
+ self.domain_count = 2
+ self.domain_specific_count = 1
+ self.setup_initial_domains()
+ self.users = {}
+
+ self.load_fixtures(default_fixtures)
+ self.create_users_across_domains()
+
+ def config_overrides(self):
+ super(DomainSpecificSQLIdentity, self).config_overrides()
+ self.config_fixture.config(
+ group='identity',
+ driver='keystone.identity.backends.ldap.Identity')
+ self.config_fixture.config(
+ group='resource',
+ driver='keystone.resource.backends.sql.Resource')
+ self.config_fixture.config(
+ group='assignment',
+ driver='keystone.assignment.backends.sql.Assignment')
+
+ def get_config(self, domain_id):
+ if domain_id == CONF.identity.default_domain_id:
+ return CONF
+ else:
+ return self.identity_api.domain_configs.get_domain_conf(domain_id)
+
+ def reload_backends(self, domain_id):
+ if domain_id == CONF.identity.default_domain_id:
+ self.load_backends()
+ else:
+ # Just reload the driver for this domain - which will pickup
+ # any updated cfg
+ self.identity_api.domain_configs.reload_domain_driver(domain_id)
+
+ def test_default_sql_plus_sql_specific_driver_fails(self):
+ # First confirm that if ldap is default driver, domain1 can be
+ # loaded as sql
+ self.config_fixture.config(
+ group='identity',
+ driver='keystone.identity.backends.ldap.Identity')
+ self.config_fixture.config(
+ group='assignment',
+ driver='keystone.assignment.backends.sql.Assignment')
+ self.load_backends()
+ # Make any identity call to initiate the lazy loading of configs
+ self.identity_api.list_users(
+ domain_scope=CONF.identity.default_domain_id)
+ self.assertIsNotNone(self.get_config(self.domains['domain1']['id']))
+
+ # Now re-initialize, but with sql as the default identity driver
+ self.config_fixture.config(
+ group='identity',
+ driver='keystone.identity.backends.sql.Identity')
+ self.config_fixture.config(
+ group='assignment',
+ driver='keystone.assignment.backends.sql.Assignment')
+ self.load_backends()
+ # Make any identity call to initiate the lazy loading of configs, which
+ # should fail since we would now have two sql drivers.
+ self.assertRaises(exception.MultipleSQLDriversInConfig,
+ self.identity_api.list_users,
+ domain_scope=CONF.identity.default_domain_id)
+
+ def test_multiple_sql_specific_drivers_fails(self):
+ self.config_fixture.config(
+ group='identity',
+ driver='keystone.identity.backends.ldap.Identity')
+ self.config_fixture.config(
+ group='assignment',
+ driver='keystone.assignment.backends.sql.Assignment')
+ self.load_backends()
+ # Ensure default, domain1 and domain2 exist
+ self.domain_count = 3
+ self.setup_initial_domains()
+ # Make any identity call to initiate the lazy loading of configs
+ self.identity_api.list_users(
+ domain_scope=CONF.identity.default_domain_id)
+ # This will only load domain1, since the domain2 config file is
+ # not stored in the same location
+ self.assertIsNotNone(self.get_config(self.domains['domain1']['id']))
+
+ # Now try and manually load a 2nd sql specific driver, for domain2,
+ # which should fail.
+ self.assertRaises(
+ exception.MultipleSQLDriversInConfig,
+ self.identity_api.domain_configs._load_config_from_file,
+ self.resource_api,
+ [tests.TESTCONF + '/domain_configs_one_extra_sql/' +
+ 'keystone.domain2.conf'],
+ 'domain2')
+
+
+class LdapFilterTests(test_backend.FilterTests, tests.TestCase):
+
+ def setUp(self):
+ super(LdapFilterTests, self).setUp()
+ self.useFixture(database.Database())
+ self.clear_database()
+
+ common_ldap.register_handler('fake://', fakeldap.FakeLdap)
+ self.load_backends()
+ self.load_fixtures(default_fixtures)
+
+ self.engine = sql.get_engine()
+ self.addCleanup(sql.cleanup)
+ sql.ModelBase.metadata.create_all(bind=self.engine)
+
+ self.addCleanup(sql.ModelBase.metadata.drop_all, bind=self.engine)
+ self.addCleanup(common_ldap_core._HANDLERS.clear)
+
+ def config_overrides(self):
+ super(LdapFilterTests, self).config_overrides()
+ self.config_fixture.config(
+ group='identity',
+ driver='keystone.identity.backends.ldap.Identity')
+
+ def config_files(self):
+ config_files = super(LdapFilterTests, self).config_files()
+ config_files.append(tests.dirs.tests_conf('backend_ldap.conf'))
+ return config_files
+
+ def clear_database(self):
+ for shelf in fakeldap.FakeShelves:
+ fakeldap.FakeShelves[shelf].clear()