ungleich-user/dal/ungleich_ldap.py

205 lines
7.4 KiB
Python
Raw Normal View History

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