<NME> ldap_utils.py
<BEF> """
A collection of LDAP based classes/functions.

For a first version, the LDAP support offered by Modoboa only supports
one global server definition : the one the django-auth-ldap uses.

Extra information about Active Directory
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Extracted from `this blog
<http://www.dirmgr.com/blog/2010/8/26/ldap-password-changes-in-active-directory.html>`_:

- Active Directory doesn't appear to support the password modify
  extended operation, so you must change passwords using a normal LDAP
  modify operation.

- Active Directory stores passwords in the unicodePwd attribute,
  rather than userPassword.

- Active Directory will only accept password changes over secure
  connections. I have only ever used SSL. It may be that you can also
  use StartTLS, or perhaps SASL with confidentiality, but I'm not sure
  about that.

- The new password must be enclosed in quotation marks, and it must
  use a UTF-16 little-endian encoding.

- Active Directory may impose some strength requirements on the
  password, although exactly what those requirements are may vary from
  one instance to another.

"""

import ldap

from django.conf import settings
from django.utils.encoding import force_bytes, force_str
from django.utils.translation import ugettext as _

from django.conf import settings
from django.utils import six
from django.utils.encoding import force_bytes
from django.utils.translation import ugettext as _

from modoboa.core.password_hashers import get_password_hasher
    """LDAP authentication backend."""

    def __init__(self):
        param_tools.apply_to_django_settings()
        self.global_params = dict(param_tools.get_global_parameters("core"))
        self.server_uri = self._setting(
            "AUTH_LDAP_SERVER_URI", "ldap://localhost"
        )
        self.pwd_attr = self._setting("LDAP_PASSWORD_ATTR", "userPassword")
        self.ldap_ad = self._setting("LDAP_ACTIVE_DIRECTORY", False)
        self.conn = None
        self.user_filter = self._setting("LDAP_USER_FILTER", "")

    def _setting(self, name, default):
        try:
            value = getattr(settings, name)
        except AttributeError:
            value = default
        return value

    def _get_conn(self, dn, password):
        """Get a connection from the server."""
        conn = ldap.initialize(self.server_uri)
        conn.set_option(ldap.OPT_X_TLS_DEMAND, True)

    def _get_conn(self, dn, password):
        """Get a connection from the server."""
        kwargs = {}
        if six.PY3:
            kwargs["bytes_mode"] = False
        conn = ldap.initialize(self.server_uri, **kwargs)
        conn.set_option(ldap.OPT_X_TLS_DEMAND, True)
        conn.set_option(ldap.OPT_DEBUG_LEVEL, 255)
        conn.simple_bind_s(dn, password)
        return conn

    def connect_to_server(self, user, password):
            bind_dn = self._setting("AUTH_LDAP_BIND_DN", "")
            bind_pwd = self._setting("AUTH_LDAP_BIND_PASSWORD", "")
            self.conn = self._get_conn(bind_dn, bind_pwd)
        else:
            tpl = self._setting("AUTH_LDAP_USER_DN_TEMPLATE", "")
            self.user_dn = tpl % {"user": user}
            self.conn = self._get_conn(self.user_dn, password)

    def _find_user_dn(self, user):
        """Find the DN of the given user."""
        sbase = self.global_params["ldap_search_base"]
        sfilter = self.global_params["ldap_search_filter"]
        sfilter = sfilter % {"user": user}
        res = self.conn.search_s(
            force_str(sbase), ldap.SCOPE_SUBTREE, force_str(sfilter)
        sbase = self.global_params["ldap_search_base"]
        sfilter = self.global_params["ldap_search_filter"]
        sfilter = sfilter % {"user": user}
        res = self.conn.search_s(sbase, ldap.SCOPE_SUBTREE, sfilter)
        try:
            dn = res[0][0]
        except IndexError:

        :param clearpassword: the clear password
        :return: the encrypted password
        """
        scheme = self.global_params["password_scheme"]
        hasher = get_password_hasher(scheme.upper())("ldap")
        return hasher.encrypt(clearpassword)

    def update_user_password(self, user, password, newpassword):
        """Update user password."""
        self.connect_to_server(user, password)
        user_dn = self._find_user_dn(user)
        if self.ldap_ad:
            newpassword = (
                ('"%s"' % newpassword).encode("utf-16").lstrip("\377\376")
            )
        ldif = [(ldap.MOD_REPLACE,
                 force_str(self.pwd_attr),
                 force_bytes(self._crypt_password(newpassword)))]
                ('"%s"' % newpassword).encode("utf-16").lstrip("\377\376")
            )
        ldif = [(ldap.MOD_REPLACE,
                 self.pwd_attr,
                 force_bytes(self._crypt_password(newpassword)))]
        try:
            self.conn.modify_s(user_dn, ldif)
        except ldap.LDAPError as e:
            raise InternalError(
                _("Failed to update password: {}").format(e))

<MSG> when using LDAP use str (bytes) on PY2 and str on PY3

<DFF> @@ -39,7 +39,7 @@ import ldap
 
 from django.conf import settings
 from django.utils import six
-from django.utils.encoding import force_bytes
+from django.utils.encoding import force_bytes, force_str
 from django.utils.translation import ugettext as _
 
 from modoboa.core.password_hashers import get_password_hasher
@@ -70,13 +70,12 @@ class LDAPAuthBackend(object):
 
     def _get_conn(self, dn, password):
         """Get a connection from the server."""
-        kwargs = {}
-        if six.PY3:
-            kwargs["bytes_mode"] = False
-        conn = ldap.initialize(self.server_uri, **kwargs)
+        conn = ldap.initialize(self.server_uri, bytes_mode=six.PY2)
         conn.set_option(ldap.OPT_X_TLS_DEMAND, True)
         conn.set_option(ldap.OPT_DEBUG_LEVEL, 255)
-        conn.simple_bind_s(dn, password)
+        conn.simple_bind_s(
+            force_str(dn), force_str(password)
+        )
         return conn
 
     def connect_to_server(self, user, password):
@@ -98,7 +97,9 @@ class LDAPAuthBackend(object):
         sbase = self.global_params["ldap_search_base"]
         sfilter = self.global_params["ldap_search_filter"]
         sfilter = sfilter % {"user": user}
-        res = self.conn.search_s(sbase, ldap.SCOPE_SUBTREE, sfilter)
+        res = self.conn.search_s(
+            force_str(sbase), ldap.SCOPE_SUBTREE, force_str(sfilter)
+        )
         try:
             dn = res[0][0]
         except IndexError:
@@ -124,10 +125,10 @@ class LDAPAuthBackend(object):
                 ('"%s"' % newpassword).encode("utf-16").lstrip("\377\376")
             )
         ldif = [(ldap.MOD_REPLACE,
-                 self.pwd_attr,
+                 force_str(self.pwd_attr),
                  force_bytes(self._crypt_password(newpassword)))]
         try:
-            self.conn.modify_s(user_dn, ldif)
+            self.conn.modify_s(force_str(user_dn), ldif)
         except ldap.LDAPError as e:
             raise InternalError(
                 _("Failed to update password: {}").format(e))

when using LDAP use str (bytes) on PY2 and str on PY3