AC-156. Implement LDAP organization mapping, update settings files and comments on LDAP configuration.

awx/main/backend.py

@@ -1,8 +1,19 @@
 # Copyright (c) 2013 AnsibleWorks, Inc.
 # All Rights Reserved.
 
+# Django
+from django.dispatch import receiver
+
 # django-auth-ldap
+from django_auth_ldap.backend import LDAPSettings as BaseLDAPSettings
 from django_auth_ldap.backend import LDAPBackend as BaseLDAPBackend
+from django_auth_ldap.backend import populate_user
+
+class LDAPSettings(BaseLDAPSettings):
+
+    defaults = dict(BaseLDAPSettings.defaults.items() + {
+        'ORGANIZATION_MAP': {},
+    }.items())
 
 class LDAPBackend(BaseLDAPBackend):
     '''
@@ -11,6 +22,16 @@ class LDAPBackend(BaseLDAPBackend):
 
     settings_prefix = 'AUTH_LDAP_'
 
+    def _get_settings(self):
+        if self._settings is None:
+            self._settings = LDAPSettings(self.settings_prefix)
+        return self._settings
+
+    def _set_settings(self, settings):
+        self._settings = settings
+
+    settings = property(_get_settings, _set_settings)
+
     def authenticate(self, username, password):
         if not self.settings.SERVER_URI:
             return None
@@ -34,3 +55,51 @@ class LDAPBackend(BaseLDAPBackend):
 
     def get_group_permissions(self, user, obj=None):
         return set()
+
+def _update_m2m_from_groups(user, ldap_user, rel, opts, remove=False):
+    '''
+    Hepler function to update m2m relationship based on LDAP group membership.
+    '''
+    should_add = False
+    if opts is None:
+        return
+    elif not opts:
+        pass
+    elif opts is True:
+        should_add = True
+    else:
+        if isinstance(opts, basestring):
+            opts = [opts]
+        for group_dn in opts:
+            if not isinstance(group_dn, basestring):
+                continue
+            if ldap_user._get_groups().is_member_of(group_dn):
+                should_add = True
+    if should_add:
+        rel.add(user)
+    elif remove:
+        rel.remove(user)
+
+@receiver(populate_user)
+def on_populate_user(sender, **kwargs):
+    '''
+    Handle signal from LDAP backend to populate the user object.  Update user's
+    organization membership according to their LDAP groups.
+    '''
+    from awx.main.models import Organization
+    user = kwargs['user']
+    ldap_user = kwargs['ldap_user']
+    backend = ldap_user.backend
+
+    org_map = getattr(backend.settings, 'ORGANIZATION_MAP', {})
+    for org_name, org_opts in org_map.items():
+        org, created = Organization.objects.get_or_create(name=org_name)
+        remove = bool(org_opts.get('remove', False))
+        admins_opts = org_opts.get('admins', None)
+        remove_admins = bool(org_opts.get('remove_admins', remove))
+        _update_m2m_from_groups(user, ldap_user, org.admins, admins_opts,
+                                remove_admins)
+        users_opts = org_opts.get('users', None)
+        remove_users = bool(org_opts.get('remove_users', remove))
+        _update_m2m_from_groups(user, ldap_user, org.users, users_opts,
+                                remove_users)

awx/main/tests/base.py

@@ -18,11 +18,9 @@ from django.contrib.auth.models import User
 import django.test
 from django.test.client import Client
 
-# Django-Auth-LDAP
-from django_auth_ldap.backend import LDAPSettings
-
 # AWX
 from awx.main.models import *
+from awx.main.backend import LDAPSettings
 
 class BaseTestMixin(object):
     '''

awx/main/tests/users.py

@@ -8,7 +8,7 @@ import urllib
 
 # Django
 from django.conf import settings
-from django.contrib.auth.models import User
+from django.contrib.auth.models import User, Group
 import django.test
 from django.test.client import Client
 from django.core.urlresolvers import reverse
@@ -627,9 +627,10 @@ class UsersTest(BaseTest):
 
 class LdapTest(BaseTest):
 
-    def use_test_setting(self, name, default=None):
+    def use_test_setting(self, name, default=None, from_name=None):
+        from_name = from_name or name
         setattr(settings, 'AUTH_LDAP_%s' % name,
-                getattr(settings, 'TEST_AUTH_LDAP_%s' % name, default))
+                getattr(settings, 'TEST_AUTH_LDAP_%s' % from_name, default))
 
     def setUp(self):
         super(LdapTest, self).setUp()
@@ -647,15 +648,18 @@ class LdapTest(BaseTest):
             self.use_test_setting(name)
 
     def check_login(self, username=None, password=None, should_fail=False):
+        self.assertEqual(Group.objects.count(), 0)
         username = username or self.ldap_username
         password = password or self.ldap_password
         result = self.client.login(username=username, password=password)
         self.assertNotEqual(result, should_fail)
+        self.assertEqual(Group.objects.count(), 0)
         if not should_fail:
             return User.objects.get(username=username)
 
     def test_ldap_auth(self):
         self.use_test_setting('USER_SEARCH')
+        self.use_test_setting('ALWAYS_UPDATE_USER')
         self.assertEqual(User.objects.filter(username=self.ldap_username).count(), 0)
         # Test logging in, user should be created with no flags or fields set.
         user = self.check_login()
@@ -679,3 +683,65 @@ class LdapTest(BaseTest):
             user = self.check_login()
             for attr in settings.AUTH_LDAP_USER_ATTR_MAP.keys():
                 self.assertTrue(getattr(user, attr))
+        # Turn on group search fields.
+        for name in ('GROUP_SEARCH', 'GROUP_TYPE'):
+            self.use_test_setting(name)
+        # Test that user must be in required group.
+        self.use_test_setting('REQUIRE_GROUP', from_name='REQUIRE_GROUP_FAIL')
+        if settings.AUTH_LDAP_REQUIRE_GROUP:
+            user = self.check_login(should_fail=True)
+        self.use_test_setting('REQUIRE_GROUP')
+        user = self.check_login()
+        # Test that user must not be in deny group.
+        self.use_test_setting('DENY_GROUP', from_name='DENY_GROUP_FAIL')
+        if settings.AUTH_LDAP_DENY_GROUP:
+            user = self.check_login(should_fail=True)
+        self.use_test_setting('DENY_GROUP')
+        user = self.check_login()
+        # Check that user flags are set from group membership.
+        self.use_test_setting('USER_FLAGS_BY_GROUP')
+        if settings.AUTH_LDAP_USER_FLAGS_BY_GROUP:
+            user = self.check_login()
+            for attr in settings.AUTH_LDAP_USER_FLAGS_BY_GROUP.keys():
+                self.assertTrue(getattr(user, attr))
+
+    def test_ldap_organization_mapping(self):
+        for name in ('USER_SEARCH', 'ALWAYS_UPDATE_USER', 'USER_ATTR_MAP',
+                     'GROUP_SEARCH', 'GROUP_TYPE', 'USER_FLAGS_BY_GROUP'):
+            self.use_test_setting(name)
+        self.assertEqual(User.objects.filter(username=self.ldap_username).count(), 0)
+        self.use_test_setting('ORGANIZATION_MAP', {})
+        self.use_test_setting('ORGANIZATION_MAP_RESULT', {})
+        for org_name in settings.AUTH_LDAP_ORGANIZATION_MAP.keys():
+            self.assertEqual(Organization.objects.filter(name=org_name).count(), 0)
+        user = self.check_login()
+        for org_name in settings.AUTH_LDAP_ORGANIZATION_MAP.keys():
+            self.assertEqual(Organization.objects.filter(name=org_name).count(), 1)
+        for org_name, org_result in settings.AUTH_LDAP_ORGANIZATION_MAP_RESULT.items():
+            org = Organization.objects.get(name=org_name)
+            if org_result.get('admins', False):
+                self.assertTrue(user in org.admins.all())
+            else:
+                self.assertFalse(user in org.admins.all())
+            if org_result.get('users', False):
+                self.assertTrue(user in org.users.all())
+            else:
+                self.assertFalse(user in org.users.all())
+        # Try again with different test mapping.
+        self.use_test_setting('ORGANIZATION_MAP', {},
+                              from_name='ORGANIZATION_MAP_2')
+        self.use_test_setting('ORGANIZATION_MAP_RESULT', {},
+                              from_name='ORGANIZATION_MAP_2_RESULT')
+        user = self.check_login()
+        for org_name in settings.AUTH_LDAP_ORGANIZATION_MAP.keys():
+            self.assertEqual(Organization.objects.filter(name=org_name).count(), 1)
+        for org_name, org_result in settings.AUTH_LDAP_ORGANIZATION_MAP_RESULT.items():
+            org = Organization.objects.get(name=org_name)
+            if org_result.get('admins', False):
+                self.assertTrue(user in org.admins.all())
+            else:
+                self.assertFalse(user in org.admins.all())
+            if org_result.get('users', False):
+                self.assertTrue(user in org.users.all())
+            else:
+                self.assertFalse(user in org.users.all())