Configurable actions on LDAP backend in users Active Directory (bug 1…

…052929)

Change-Id: I99092eb4aee3b3b1b9cf297561577f1915c0e886

etc/keystone.conf.sample

@@ -112,17 +112,26 @@
 # user_objectclass = inetOrgPerson
 # user_id_attribute = cn
 # user_name_attribute = sn
+# user_allow_create = True
+# user_allow_update = True
+# user_allow_delete = True
 
 # tenant_tree_dn = ou=Groups,dc=example,dc=com
 # tenant_objectclass = groupOfNames
 # tenant_id_attribute = cn
 # tenant_member_attribute = member
 # tenant_name_attribute = ou
+# tenant_allow_create = True
+# tenant_allow_update = True
+# tenant_allow_delete = True
 
 # role_tree_dn = ou=Roles,dc=example,dc=com
 # role_objectclass = organizationalRole
 # role_id_attribute = cn
 # role_member_attribute = roleOccupant
+# role_allow_create = True
+# role_allow_update = True
+# role_allow_delete = True
 
 [filter:debug]
 paste.filter_factory = keystone.common.wsgi:Debug.factory

keystone/common/ldap/core.py

@@ -92,6 +92,15 @@ def __init__(self,  conf):
             self.object_class = (getattr(conf.ldap, objclass)
                                  or self.DEFAULT_OBJECTCLASS)
 
+            allow_create = '%s_allow_create' % self.options_name
+            self.allow_create = getattr(conf.ldap, allow_create)
+
+            allow_update = '%s_allow_update' % self.options_name
+            self.allow_update = getattr(conf.ldap, allow_update)
+
+            allow_delete = '%s_allow_delete' % self.options_name
+            self.allow_delete = getattr(conf.ldap, allow_delete)
+
             self.structural_classes = self.DEFAULT_STRUCTURAL_CLASSES
         self.use_dumb_member = getattr(conf.ldap, 'use_dumb_member') or True
 
@@ -163,6 +172,10 @@ def affirm_unique(self, values):
                                                  values['id'])
 
     def create(self, values):
+        if not self.allow_create:
+            msg = 'LDAP backend does not allow %s create' % self.options_name
+            raise exception.ForbiddenAction(msg)
+
         conn = self.get_connection()
         object_classes = self.structural_classes + [self.object_class]
         attrs = [('objectClass', object_classes)]
@@ -262,6 +275,10 @@ def _get_page_markers(marker, limit, lst, key=lambda x: x.id):
         return (prv, nxt)
 
     def update(self, id, values, old_obj=None):
+        if not self.allow_update:
+            msg = 'LDAP backend does not allow %s update' % self.options_name
+            raise exception.ForbiddenAction(msg)
+
         if old_obj is None:
             old_obj = self.get(id)
 
@@ -285,6 +302,10 @@ def update(self, id, values, old_obj=None):
         conn.modify_s(self._id_to_dn(id), modlist)
 
     def delete(self, id):
+        if not self.allow_delete:
+            msg = 'LDAP backend does not allow %s delete' % self.options_name
+            raise exception.ForbiddenAction(msg)
+
         conn = self.get_connection()
         conn.delete_s(self._id_to_dn(id))
 

keystone/config.py

@@ -164,21 +164,29 @@ def register_cli_int(*args, **kw):
 register_bool('use_dumb_member', group='ldap', default=False)
 register_str('user_name_attribute', group='ldap', default='sn')
 
-
 register_str('user_tree_dn', group='ldap', default=None)
 register_str('user_objectclass', group='ldap', default='inetOrgPerson')
 register_str('user_id_attribute', group='ldap', default='cn')
+register_bool('user_allow_create', group='ldap', default=True)
+register_bool('user_allow_update', group='ldap', default=True)
+register_bool('user_allow_delete', group='ldap', default=True)
 
 register_str('tenant_tree_dn', group='ldap', default=None)
 register_str('tenant_objectclass', group='ldap', default='groupOfNames')
 register_str('tenant_id_attribute', group='ldap', default='cn')
 register_str('tenant_member_attribute', group='ldap', default='member')
 register_str('tenant_name_attribute', group='ldap', default='ou')
+register_bool('tenant_allow_create', group='ldap', default=True)
+register_bool('tenant_allow_update', group='ldap', default=True)
+register_bool('tenant_allow_delete', group='ldap', default=True)
 
 register_str('role_tree_dn', group='ldap', default=None)
 register_str('role_objectclass', group='ldap', default='organizationalRole')
 register_str('role_id_attribute', group='ldap', default='cn')
 register_str('role_member_attribute', group='ldap', default='roleOccupant')
+register_bool('role_allow_create', group='ldap', default=True)
+register_bool('role_allow_update', group='ldap', default=True)
+register_bool('role_allow_delete', group='ldap', default=True)
 
 #pam
 register_str('url', group='pam', default=None)

keystone/identity/backends/ldap/core.py

@@ -947,6 +947,6 @@ def update(self, role_id, role):
         except exception.NotFound:
             pass
         try:
-            super(RoleApi, self).update(id, role)
+            super(RoleApi, self).update(role_id, role)
         except exception.NotFound:
-            raise exception.UserNotFound(user_id=id)
+            raise exception.RoleNotFound(role_id=role_id)

tests/test_backend_ldap.py

@@ -65,3 +65,140 @@ def test_build_tree(self):
         user_api = identity_ldap.UserApi(CONF)
         self.assertTrue(user_api)
         self.assertEquals(user_api.tree_dn, "ou=Users,%s" % CONF.ldap.suffix)
+
+    def test_configurable_allowed_user_actions(self):
+        self.config([test.etcdir('keystone.conf.sample'),
+                     test.testsdir('test_overrides.conf'),
+                     test.testsdir('backend_ldap.conf')])
+        self.identity_api = identity_ldap.Identity()
+
+        user = {'id': 'fake1',
+                'name': 'fake1',
+                'password': 'fakepass1',
+                'tenants': ['bar']}
+        self.identity_api.create_user('fake1', user)
+        user_ref = self.identity_api.get_user('fake1')
+        self.assertEqual(user_ref['id'], 'fake1')
+
+        user['password'] = 'fakepass2'
+        self.identity_api.update_user('fake1', user)
+
+        self.identity_api.delete_user('fake1')
+        self.assertRaises(exception.UserNotFound,
+                          self.identity_api.get_user,
+                          'fake1')
+
+    def test_configurable_forbidden_user_actions(self):
+        self.config([test.etcdir('keystone.conf.sample'),
+                     test.testsdir('test_overrides.conf'),
+                     test.testsdir('backend_ldap.conf')])
+        CONF.ldap.user_allow_create = False
+        CONF.ldap.user_allow_update = False
+        CONF.ldap.user_allow_delete = False
+        self.identity_api = identity_ldap.Identity()
+
+        user = {'id': 'fake1',
+                'name': 'fake1',
+                'password': 'fakepass1',
+                'tenants': ['bar']}
+        self.assertRaises(exception.ForbiddenAction,
+                          self.identity_api.create_user,
+                          'fake1',
+                          user)
+
+        self.user_foo['password'] = 'fakepass2'
+        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_allowed_tenant_actions(self):
+        self.config([test.etcdir('keystone.conf.sample'),
+                     test.testsdir('test_overrides.conf'),
+                     test.testsdir('backend_ldap.conf')])
+        self.identity_api = identity_ldap.Identity()
+
+        tenant = {'id': 'fake1', 'name': 'fake1'}
+        self.identity_api.create_tenant('fake1', tenant)
+        tenant_ref = self.identity_api.get_tenant('fake1')
+        self.assertEqual(tenant_ref['id'], 'fake1')
+
+        tenant['enabled'] = 'False'
+        self.identity_api.update_tenant('fake1', tenant)
+
+        self.identity_api.delete_tenant('fake1')
+        self.assertRaises(exception.TenantNotFound,
+                          self.identity_api.get_tenant,
+                          'fake1')
+
+    def test_configurable_forbidden_tenant_actions(self):
+        self.config([test.etcdir('keystone.conf.sample'),
+                     test.testsdir('test_overrides.conf'),
+                     test.testsdir('backend_ldap.conf')])
+        CONF.ldap.tenant_allow_create = False
+        CONF.ldap.tenant_allow_update = False
+        CONF.ldap.tenant_allow_delete = False
+        self.identity_api = identity_ldap.Identity()
+
+        tenant = {'id': 'fake1', 'name': 'fake1'}
+        self.assertRaises(exception.ForbiddenAction,
+                          self.identity_api.create_tenant,
+                          'fake1',
+                          tenant)
+
+        self.tenant_bar['enabled'] = 'False'
+        self.assertRaises(exception.ForbiddenAction,
+                          self.identity_api.update_tenant,
+                          self.tenant_bar['id'],
+                          self.tenant_bar)
+        self.assertRaises(exception.ForbiddenAction,
+                          self.identity_api.delete_tenant,
+                          self.tenant_bar['id'])
+
+    def test_configurable_allowed_role_actions(self):
+        self.config([test.etcdir('keystone.conf.sample'),
+                     test.testsdir('test_overrides.conf'),
+                     test.testsdir('backend_ldap.conf')])
+        self.identity_api = identity_ldap.Identity()
+
+        role = {'id': 'fake1', 'name': 'fake1'}
+        self.identity_api.create_role('fake1', role)
+        role_ref = self.identity_api.get_role('fake1')
+        self.assertEqual(role_ref['id'], 'fake1')
+
+        role['name'] = 'fake2'
+        self.identity_api.update_role('fake1', role)
+
+        self.identity_api.delete_role('fake1')
+        self.assertRaises(exception.RoleNotFound,
+                          self.identity_api.get_role,
+                          'fake1')
+
+    def test_configurable_forbidden_role_actions(self):
+        self.config([test.etcdir('keystone.conf.sample'),
+                     test.testsdir('test_overrides.conf'),
+                     test.testsdir('backend_ldap.conf')])
+        CONF.ldap.role_allow_create = False
+        CONF.ldap.role_allow_update = False
+        CONF.ldap.role_allow_delete = False
+        self.identity_api = identity_ldap.Identity()
+
+        role = {'id': 'fake1', 'name': 'fake1'}
+        self.assertRaises(exception.ForbiddenAction,
+                          self.identity_api.create_role,
+                          'fake1',
+                          role)
+
+        self.role_useless['name'] = 'useful'
+        self.assertRaises(exception.ForbiddenAction,
+                          self.identity_api.update_role,
+                          self.role_useless['id'],
+                          self.role_useless)
+
+        self.assertRaises(exception.ForbiddenAction,
+                          self.identity_api.delete_role,
+                          self.role_useless['id'])