Rewriting to use ldap3

.travis.yml

@@ -1,24 +1,18 @@
 language: python
 
-
 python:
   - "2.7"
+  - "3.4"
 
 env:
-  - DJANGO_VERSION=1.4
   - DJANGO_VERSION=1.8
-
-apt:
-  packages:
-    - libldap2-dev
-    - libsasl2-dev
+  - DJANGO_VERSION=1.11
 
 sudo: false
 
 install: 
- - pip install Django==$DJANGO_VERSION
+ - pip install Django~=$DJANGO_VERSION
  - pip install .
- - pip install -r testing-ci/requirements.txt 
  - pip install pep8 pyflakes
 
 before_script:

django_auth_ldap_ad/backend.py

@@ -1,7 +1,6 @@
-
-import ldap
-import ldap.sasl
+import ldap3
 from django.contrib.auth.models import User, Group
+from ldap3.core import exceptions
 
 import six
 
@@ -16,41 +15,43 @@ class LDAPBackendException(Exception):
 
 class LDAPBackend(object):
 
-    def authenticate(self, username=None, password=None):
-        if not hasattr(self, "ldap_settings"):
-            self.ldap_settings = LDAPSettings()
+    def __init__(self):
+        # References are stored as instance variables so tests can replace
+        # these in the instance with fakes/mocks without affecting
+        # business logic
+        self.connection = ldap3.Connection
+        self.ldap_settings = LDAPSettings()
 
+    def _generate_servers(self):
         if isinstance(self.ldap_settings.SERVER_URI, six.string_types):
-            servers_urls = [self.ldap_settings.SERVER_URI]
+            server_urls = [self.ldap_settings.SERVER_URI]
         else:
-            servers_urls = self.ldap_settings.SERVER_URI
-
-        # For all configured servers try to connect
-        for server in servers_urls:
-            # Use self.ldap_connection object if such is given for testing with
-            # mockldap.
-            if not hasattr(self, "ldap_connection"):
-                try:
-                    ldap_connection = self.ldap_open_connection(
-                        server, username, password)
-                except ldap.SERVER_DOWN:
-                    continue
-                except ldap.INVALID_CREDENTIALS:
-                    return None
-            else:  # end up here with mock
-                ldap_connection = self.ldap_connection
+            server_urls = self.ldap_settings.SERVER_URI
+        for server_url in server_urls:
+            yield ldap3.Server(server_url)
 
-            for key, value in self.ldap_settings.CONNECTION_OPTIONS.items():
-                ldap_connection.set_option(key, value)
+    def authenticate(self, username=None, password=None):
 
-            # Do search
+        # For all configured servers try to connect
+        for server in self._generate_servers():
             try:
-                ldap_user_info = self.ldap_search_user(
-                    ldap_connection, username, password)
-            except LDAPBackendException:
+                ldap_connection = self.ldap_open_connection(
+                    server, username, password)
+            except exceptions.LDAPSocketOpenError:
+                continue
+            except exceptions.LDAPInvalidCredentialsResult:
                 return None
+            try:
+                # Do search
+                try:
+                    ldap_user_info = self.ldap_search_user(
+                        ldap_connection, username, password)
+                except LDAPBackendException:
+                    return None
 
-            return self.get_local_user(username, ldap_user_info)
+                return self.get_local_user(username, ldap_user_info)
+            finally:
+                ldap_connection.unbind()
         return None
 
     def get_user(self, user_id):
@@ -59,32 +60,30 @@ def get_user(self, user_id):
         except User.DoesNotExist:
             return None
 
-    def ldap_open_connection(self, ldap_url, username, password):
-        ldap_session = ldap.initialize(
-            ldap_url, trace_level=self.ldap_settings.TRACE_LEVEL)
-
-        sasl_auth = ldap.sasl.sasl({
-            ldap.sasl.CB_AUTHNAME: username,
-            ldap.sasl.CB_PASS: password,
-        },
-            self.ldap_settings.SASL_MECH
-        )
-
-        ldap_session.sasl_interactive_bind_s("", sasl_auth)
-        return ldap_session
+    def ldap_open_connection(self, server, username, password):
+        kwargs = {
+            "server": server, "user": username, "password": password,
+            "authentication": ldap3.SASL,
+            "sasl_mechanism": self.ldap_settings.SASL_MECH
+        }
+        kwargs.update(self.ldap_settings.CONNECTION_OPTIONS)
+        kwargs["client_strategy"] = ldap3.SYNC
+        connection = self.connection(**kwargs)
+        connection.bind()
+        return connection
 
     # Search for user, returns users info (dict)
     def ldap_search_user(self, connection, username, password):
-        ldap_result_id = connection.search(
-            self.ldap_settings.SEARCH_DN,
-            ldap.SCOPE_SUBTREE,
-            self.ldap_settings.SEARCH_FILTER % {
-                "user": username})
-        result_all_type, result_all_data = connection.result(ldap_result_id, 1)
-        result_entries = []
-        for result_type, result_data in result_all_data:
-            if result_type is not None:
-                result_entries.append(result_data)
+        attributes = ["memberOf"] + list(self.ldap_settings.USER_ATTR_MAP.values())
+        if not connection.search(
+            search_base=self.ldap_settings.SEARCH_DN,
+            search_scope=ldap3.SUBTREE,
+            attributes=attributes,
+            search_filter=self.ldap_settings.SEARCH_FILTER % {
+                "user": username}):
+            raise LDAPBackendException("Failure searching for user")
+
+        result_entries = connection.entries
 
         if len(result_entries) == 0:
             raise LDAPBackendException("No entries found!")
@@ -104,7 +103,11 @@ def get_local_user(self, ldap_username, info):
             user.set_unusable_password()
         # refresh memberships
         members_of = []
-        for group in info.get('memberOf', []):
+        try:
+            groups = info["memberOf"]
+        except AttributeError:
+            groups = []
+        for group in groups:
             members_of.append(group.lower().split(","))
 
         # Set first_name or last_name or email ..
@@ -154,7 +157,7 @@ def check_for_membership(members_of, required_groups_options):
 
 class LDAPSettings(object):
     defaults = {
-        'CONNECTION_OPTIONS': {ldap.OPT_REFERRALS: 0},
+        'CONNECTION_OPTIONS': {},
         'SERVER_URI': 'ldap://localhost',
         'USER_FLAGS_BY_GROUP': {},
         'USER_GROUPS_BY_GROUP': {},

django_auth_ldap_ad/tests.py

@@ -1,11 +1,9 @@
-
-
-import mockldap
-import backend
+from __future__ import absolute_import
+import ldap3
+from . import backend
 
 from django.test import TestCase
 
-
 from django.contrib.auth.models import User, Group
 
 
@@ -26,41 +24,28 @@ class LDAPBackendTest(TestCase):
                            "dc=test,cn=superuser,cn=extra,cn=fake,ou=foo",
                            "dc=test,cn=fakuser,cn=extra,cn=fake,ou=foo"]})
 
-    # This is the content of our mock LDAP directory. It takes the form
-    # {dn: {attr: [value, ...], ...}, ...}.
-    directory = dict([top, alice])
-
-    @classmethod
-    def setUpClass(cls):
-        # We only need to create the MockLdap instance once. The content we
-        # pass in will be used for all LDAP connections.
-        cls.mockldap = mockldap.MockLdap(cls.directory)
-
-    @classmethod
-    def tearDownClass(cls):
-        del cls.mockldap
-
     def setUp(self):
-        self.mockldap.start()
-        self.ldapobj = self.mockldap['ldap://localhost/']
         self.backend = backend.LDAPBackend()
-        self.backend.ldap_connection = self.ldapobj
+        self.backend.connection = self._create_connection
+        self.last_connection = None
 
-    def tearDown(self):
-        # Stop patching ldap.initialize and reset state.
-        self.mockldap.stop()
-        del self.ldapobj
+    def _create_connection(self, **kwargs):
+        kwargs["client_strategy"] = ldap3.MOCK_SYNC
+        connection = ldap3.Connection(**kwargs)
+        connection.strategy.add_entry(*self.alice)
+        self.last_connection = connection
+        return connection
 
     def _init_settings(self, **kwargs):
         self.backend.ldap_settings = TestSettings(**kwargs)
 
     def test_options(self):
         self._init_settings(
             SEARCH_DN="o=test",
-            CONNECTION_OPTIONS={'opt1': 'value1'}
+            CONNECTION_OPTIONS={'read_only': True}
         )
         self.backend.authenticate(username='alice', password='alicepw')
-        self.assertEqual(self.ldapobj.get_option('opt1'), 'value1')
+        self.assertEqual(self.last_connection.read_only, True)
 
     def test_server_uri_string(self):
         self._init_settings(

setup.py

@@ -13,5 +13,5 @@
     version = "2.0.0",
     packages = ["django_auth_ldap_ad"],
     long_description = long_description,
-    install_requires = ["python-ldap", "Django>=1.4,<=1.9", "six"]
+    install_requires = ["ldap3", "Django>=1.8,<1.12", "six"]
 )