dynamicweb/utils/ldap_manager.py

import base64
import hashlib
import random
import ldap3
import logging
import unicodedata

from django.conf import settings

logger = logging.getLogger(__name__)


class LdapManager:
    __instance = None

    def __new__(cls):
        if LdapManager.__instance is None:
            LdapManager.__instance = object.__new__(cls)
        return LdapManager.__instance

    def __init__(self):
        """
        Initialize the LDAP subsystem.
        """
        self.rng = random.SystemRandom()
        self.server = ldap3.Server(settings.AUTH_LDAP_SERVER)

    def get_admin_conn(self):
        """
        Return a bound :class:`ldap3.Connection` instance which has write
        permissions on the dn in which the user accounts reside.
        """
        conn = self.get_conn(user=settings.LDAP_ADMIN_DN,
                             password=settings.LDAP_ADMIN_PASSWORD,
                             raise_exceptions=True)
        conn.bind()
        return conn

    def get_conn(self, **kwargs):
        """
        Return an unbound :class:`ldap3.Connection` which talks to the configured
        LDAP server.

        The *kwargs* are passed to the constructor of :class:`ldap3.Connection` and
        can be used to set *user*, *password* and other useful arguments.
        """
        return ldap3.Connection(self.server, **kwargs)

    def _ssha_password(self, password):
        """
        Apply the SSHA password hashing scheme to the given *password*.
        *password* must be a :class:`bytes` object, containing the utf-8
        encoded password.

        Return a :class:`bytes` object containing ``ascii``-compatible data
        which can be used as LDAP value, e.g. after armoring it once more using
        base64 or decoding it to unicode from ``ascii``.
        """
        SALT_BYTES = 15

        sha1 = hashlib.sha1()
        salt = self.rng.getrandbits(SALT_BYTES * 8).to_bytes(SALT_BYTES, "little")
        sha1.update(password)
        sha1.update(salt)

        digest = sha1.digest()
        passwd = b"{SSHA}" + base64.b64encode(digest + salt)
        return passwd

    def create_user(self, user, password, firstname, lastname, email):
        conn = self.get_admin_conn()
        uidNumber = self._get_max_uid() + 1
        logger.debug("uidNumber={uidNumber}".format(uidNumber=uidNumber))
        user_exists = True
        while user_exists:
            user_exists, _ = self.check_user_exists(
                "",
                '(&(objectClass=inetOrgPerson)(objectClass=posixAccount)'
                '(objectClass=top)(uidNumber={uidNumber}))'.format(
                    uidNumber=uidNumber
                )
            )
            if user_exists:
                logger.debug(
                    "{uid} exists. Trying next.".format(uid=uidNumber)
                )
                uidNumber += 1
        logger.debug("{uid} does not exist. Using it".format(uid=uidNumber))
        self._set_max_uid(uidNumber)
        try:
            uid = user
            conn.add("uid={uid},{customer_dn}".format(
                    uid=uid, customer_dn=settings.LDAP_CUSTOMER_DN
                ),
                ["inetOrgPerson", "posixAccount", "ldapPublickey"],
                {
                    "uid": [uid],
                    "sn": [lastname.encode("utf-8")],
                    "givenName": [firstname.encode("utf-8")],
                    "cn": [uid],
                    "displayName": ["{} {}".format(firstname, lastname).encode("utf-8")],
                    "uidNumber": [str(uidNumber)],
                    "gidNumber": [str(settings.LDAP_CUSTOMER_GROUP_ID)],
                    "loginShell": ["/bin/bash"],
                    "homeDirectory": ["/home/{}".format(unicodedata.normalize('NFKD', user).encode('ascii','ignore'))],
                    "mail": email.encode("utf-8"),
                    "userPassword": [self._ssha_password(
                        password.encode("utf-8")
                    )]
                }
            )
            logger.debug('Created user %s %s' % (user.encode('utf-8'),
                                                 uidNumber))
        except Exception as ex:
            logger.debug('Could not create user %s' % user.encode('utf-8'))
            logger.error("Exception: " + str(ex))
            raise Exception(ex)
        finally:
            conn.unbind()

    def change_password(self, uid, new_password):
        """
        Changes the password of the user identified by user_dn

        :param uid: str The uid that identifies the user
        :param new_password: str The new password string
        :return: True if password was changed successfully False otherwise
        """
        conn = self.get_admin_conn()

        # Make sure the user exists first to change his/her details
        user_exists, entries = self.check_user_exists(
            uid=uid,
            search_base=settings.ENTIRE_SEARCH_BASE
        )
        return_val = False
        if user_exists:
            try:
                return_val = conn.modify(
                    entries[0].entry_dn,
                    {
                        "userpassword": (
                            ldap3.MODIFY_REPLACE,
                            [self._ssha_password(new_password.encode("utf-8"))]
                        )
                    }
                )
            except Exception as ex:
                logger.error("Exception: " + str(ex))
        else:
            logger.error("User {} not found".format(uid))

        conn.unbind()
        return return_val


    def change_user_details(self, uid, details):
        """
        Updates the user details as per given values in kwargs of the user
        identified by user_dn.

        Assumes that all attributes passed in kwargs are valid.

        :param uid: str The uid that identifies the user
        :param details: dict A dictionary containing the new values
        :return: True if user details were updated successfully False otherwise
        """
        conn = self.get_admin_conn()

        # Make sure the user exists first to change his/her details
        user_exists, entries = self.check_user_exists(
            uid=uid,
            search_base=settings.ENTIRE_SEARCH_BASE
        )

        return_val = False
        if user_exists:
            details_dict = {k: (ldap3.MODIFY_REPLACE, [v.encode("utf-8")]) for
                            k, v in details.items()}
            try:
                return_val = conn.modify(entries[0].entry_dn, details_dict)
                msg = "success"
            except Exception as ex:
                msg = str(ex)
                logger.error("Exception: " + msg)
            finally:
                conn.unbind()
        else:
            msg = "User {} not found".format(uid)
            logger.error(msg)
            conn.unbind()
        return return_val, msg

    def check_user_exists(self, uid, search_filter="", attributes=None,
                          search_base=settings.LDAP_CUSTOMER_DN, search_attr="uid"):
        """
        Check if the user with the given uid exists in the customer group.

        :param uid: str representing the user
        :param search_filter: str representing the filter condition to find
                            users. If its empty, the search finds the user with
                            the given uid.
        :param attributes: list A list of str representing all the attributes
                           to be obtained in the result entries
        :param search_base: str
        :return: tuple (bool, [ldap3.abstract.entry.Entry ..])
                       A bool indicating if the user exists
                       A list of all entries obtained in the search
        """
        conn = self.get_admin_conn()
        entries = []
        try:
            result = conn.search(
                search_base=search_base,
                search_filter=search_filter if len(search_filter) > 0 else
                '(uid={uid})'.format(uid=uid),
                attributes=attributes
            )
            entries = conn.entries
        finally:
            conn.unbind()
        return result, entries

    def delete_user(self, uid):
        """
        Deletes the user with the given uid from ldap

        :param uid: str representing the user
        :return: True if the delete was successful False otherwise
        """
        conn = self.get_admin_conn()
        try:
            return_val = conn.delete(
                ("uid={uid}," + settings.LDAP_CUSTOMER_DN).format(uid=uid),
            )
            msg = "success"
        except Exception as ex:
            msg = str(ex)
            logger.error("Exception: " + msg)
            return_val = False
        finally:
            conn.unbind()
        return return_val, msg

    def _set_max_uid(self, max_uid):
        """
        a utility function to save max_uid value to a file

        :param max_uid: an integer representing the max uid
        :return:
        """
        with open(settings.LDAP_MAX_UID_FILE_PATH, 'w+') as handler:
            handler.write(str(max_uid))

    def _get_max_uid(self):
        """
        A utility function to read the max uid value that was previously set

        :return: An integer representing the max uid value that was previously
         set
        """
        try:
            with open(settings.LDAP_MAX_UID_FILE_PATH, 'r+') as handler:
                try:
                    return_value = int(handler.read())
                except ValueError as ve:
                    logger.error(
                        "Error reading int value from {}. {}"
                        "Returning default value {} instead".format(
                            settings.LDAP_MAX_UID_FILE_PATH,
                            str(ve),
                            settings.LDAP_DEFAULT_START_UID
                        )
                    )
                    return_value = settings.LDAP_DEFAULT_START_UID
                return return_value
        except FileNotFoundError as fnfe:
            logger.error("File not found : " + str(fnfe))
            return_value = settings.LDAP_DEFAULT_START_UID
            logger.error("So, returning UID={}".format(return_value))
            return return_value