vuejs-userservice/dal/ungleich_ldap.py

import base64
import hashlib
import random

import ldap3
from django.conf import settings
import logging

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))
        results = True
        while results:
            results = self.check_user_exists(
                "",
                True,
                '(&(objectClass=inetOrgPerson)(objectClass=posixAccount)'
                '(objectClass=top)(uidNumber={uidNumber}))'.format(
                    uidNumber=uidNumber
                )
            )
            if results:
                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:
            conn.add(
                ("uid={uid}," + settings.LDAP_CUSTOMER_DN).format(uid=user),
                ["inetOrgPerson", "posixAccount", "ldapPublickey"],
                {
                    "uid": [user.encode("utf-8")],
                    "sn": [lastname.encode("utf-8")],
                    "givenName": [firstname.encode("utf-8")],
                    "cn": ["{} {}".format(firstname, lastname).encode("utf-8")],
                    "displayName": ["{} {}".format(firstname, lastname).encode("utf-8")],
                    "uidNumber": [str(uidNumber)],
                    "gidNumber": [str(settings.LDAP_CUSTOMER_GROUP_ID)],
                    "loginShell": ["/bin/bash"],
                    "homeDirectory": ["/home/{}".format(user).encode("utf-8")],
                    "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, user_dn, new_password):
        """
        Changes the password of the user identified by user_dn

        :param user_dn: str The distinguished name for identifying the user
        :param new_password: str The new password string
        :return: True if password was changed successfully False otherwise
        """
        conn = self.get_admin_conn()
        return_val = conn.modify(
            user_dn,
            {
                "userpassword": (
                    ldap3.MODIFY_REPLACE,
                    [self._ssha_password(new_password.encode("utf-8"))]
                )
            }
        )
        conn.unbind()
        return return_val

    def check_user_exists(self, uid, is_customer=True, search_filter=""):
        """
        Check if the user with the given uid exists in the customer group.

        :param uid: str representing the user
        :param is_customer: bool representing whether the current user is a
                            customer. By default, the user is a customer (assume)
        :param search_filter: str representing the filter condition to find
                            users. If its empty, the search finds the user with
                            the given uid.
        :return: True if the user exists otherwise return False
        """
        conn = self.get_admin_conn()
        try:
            result = conn.search(
                settings.LDAP_CUSTOMER_DN if is_customer else settings.LDAP_USERS_DN,
                search_filter=search_filter if len(search_filter)> 0 else
                '(uid={uid})'.format(uid=uid)
            )
        finally:
            conn.unbind()
        return result

    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_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