luser/luser/models.py

import ldap3
from ldap3 import Server,Connection,ALL,MODIFY_REPLACE
from datetime import datetime

OBJECTCLASSES = ['top', 'person', 'organizationalPerson', 'inetOrgPerson', 'posixAccount', 'shadowAccount']
USERATTRIBUTES = ['cn' , 'sn', 'givenName', 'uid', 'uidNumber' , 'gidNumber', 'homeDirectory', 'loginShell', 'gecos' , 'shadowLastChange', 'shadowMax', 'userPassword', 'mail', 'description']

class LUSER():
    '''
        Class that represents secure connection to LDAP server

        LDAPhost := string IP or hostname of LDAP server
        admin_user := string DN of LDAP admin user
        admin_pass := string password of LDAP admin user
        base := string base in LDAP system where users are made
    '''

    def findlastlog(self):
        '''
            Return the largest uidNumber attribute of all users in base
        '''
        self.ldapconnection.search(search_base=self.logbase,search_filter=f'(objectClass=inetOrgPerson)', attributes=['uid'])

        alllogs = self.ldapconnection.response

        max = 0

        for i in alllogs:
            i_log = i['attributes']['uid']
            if type(i_log) is list:
                i_log = i_log[0]
            if str(i_log) == 'total':
                continue
            if type(i_log) is str or type(i_log) is int:
                i_log = int(i_log)

                if i_log > max:
                    max = i_log

        return max

    def getlastlog(self)->int:
        try:
            self.ldapconnection.search(search_base=f'uid=total,{self.logbase}',search_filter='(objectClass=person)', attributes=['uidNumber'])
            response = self.ldapconnection.response
        except:
            self.setlastlog(0)

        if response == []:
            self.setlastlog(0)
            response = 0
        else:
            response = int(response[0]['attributes']['uidNumber'])
        return response

    def setlastlog(self, newvalue: int):
        newvalue = int(newvalue)

        # Check if total record already present
        self.ldapconnection.search(search_base=f'uid=total,{self.logbase}',search_filter='(objectClass=person)', attributes=['uidNumber'])
        response = self.ldapconnection.response
 
        attributes = {'cn' : 'total', 'sn' : 'total', 'givenName' : 'total', 'uid' : 'total', 'uidNumber' : newvalue, 'gidNumber' : newvalue, 'homeDirectory' : f'/home/total', 'loginShell' : '/usr/bin/git-shell', 'gecos' : 'SystemUser', 'shadowLastChange' : self.lastpwchangenow(), 'shadowMax' : '45', 'userPassword' : 'total', 'mail' : f'total@{self.domain}' }

        if response == []:
            self.ldapconnection.add(f'uid=total,{self.logbase}', OBJECTCLASSES, attributes)
        else:
            self.ldapconnection.modify(f'uid=total,{self.logbase}', {'uidNumber' : (ldap3.MODIFY_REPLACE, [newvalue])})

        return self.ldapconnection.response

    def findlastuid(self):
        '''
            Return the largest uidNumber attribute of all users in base
        '''
        self.ldapconnection.search(search_base=self.base,search_filter=f'(objectClass=inetOrgPerson)', attributes=['uidNumber'])

        alluids = self.ldapconnection.response

        max = 0

        for i in alluids:
            i_uid = i['attributes']['uidNumber']
            if type(i_uid) is str or type(i_uid) is int:
                i_uid = int(i_uid)

                if i_uid > max:
                    max = i_uid

        return max

    def expandbase(self):
        '''
            Extract orgnaization, name of dc object and full domain part with all dc values from base
        '''
        # Split base string with commas to find values of organization and dc
        baselist = self.base.split(",")
        
        organization = ''
        dc = ''
        dcfull = ''
        domain = ''

        # Find ou in base and set it as organization variable
        for i in baselist:
            if i.split('=')[0] == 'ou':
                organization = i.split('=')[1]

        # Find first dc and set it as dc variable        
        for i in baselist:
            if i.split('=')[0] == 'dc':
                dc = i.split('=')[1]
                break

        # Find full dc and set it as dcfull variable
        for i in baselist:
            if i.split('=')[0] == 'dc':
            # if first dc, add it from dc variable
                if dcfull == '' and domain == '':
                    dcfull = f'dc={dc}'
                    domain = dc
                else:
                    dcfull += ',dc=' + i.split('=')[1]
                    domain += f'.{i.split("=")[1]}'
 
        return organization, dc, dcfull, domain

    def __init__(self, ldap_host, admin_user, admin_pass, base, autoconnect=True, lastUID = 1000):
        self.ldap_host = ldap_host
        self.admin_user = admin_user
        self.admin_pass = admin_pass
        self.base = base
        self.organization, self.dc, self.dcfull, self.domain = self.expandbase()
        self.logbase = f'ou=log,{self.dcfull}'
        self.autoconnect = autoconnect
        ldapserver = Server(ldap_host, use_ssl=True)
        lastuidfound = 0
        if self.autoconnect:
            self.ldapconnection = Connection(ldapserver, admin_user, admin_pass, auto_bind=True)
            # uid and gid of most recently registered users
            lastuidfound = self.findlastuid()
            lastlogfound = self.findlastlog()
        else:
            self.ldapconnection = Connection(ldapserver, admin_user, admin_pass, auto_bind=False)

        # Check if base and log base is created
        self.ldapconnection.search(search_base=f'{self.base}',search_filter='(objectClass=organizationalUnit)', attributes=['ou'])
        if self.ldapconnection.response == []:
            self.prepare()

        self.ldapconnection.search(search_base=f'{self.logbase}',search_filter='(objectClass=organizationalUnit)', attributes=['ou'])
        if self.ldapconnection.response == []:
            self.prepare()

        if lastuidfound == 0:
            self.lastuid = lastUID
            self.lastgid = lastUID
        else:
            self.lastuid = lastuidfound 
            self.lastgid = lastuidfound

        self.lastlog = lastlogfound
        self.setlastlog(lastlogfound)

    def prepare(self):
        '''
           Create base on LDAP host
        '''

        # Create dcObject on LDAP server and store boolean indicating it's success
        rcode1 = self.ldapconnection.add(self.dcfull, ['dcObject', 'organization'], {'o' : self.dc, 'dc' : self.dc})

        # Create organizational units on LDAP server and store boolean indicating it's success
        rcode2 = self.ldapconnection.add(self.base, ['top', 'organizationalUnit'], {'ou' : self.organization})

        # Create organizational units for log on LDAP server and store boolean indicating it's success
        rcode3 = self.ldapconnection.add(self.logbase, ['top', 'organizationalUnit'], {'ou' : 'log'})

        return rcode1 and rcode2 and rcode3

    def lastpwchangenow(self):
        '''
            Return time of last password change for the user set to current time 
            messured in days from UNIX epoch
        '''

        return str((datetime.utcnow() - datetime(1970,1,1)).days)

    def add(self, user, password):
        '''
            Add a user to base in LDAP with user and pass as credentials
            user := string containing username
            password := string containing user password
        '''
        # Increase UID and GID counters
        self.lastuid += 1
        self.lastgid += 1

        # Add user to base 
        id = f"uid={user}"
        lastlog = self.lastlog + 1

        # Object classes of a user entry
        objectClass = ['top', 'person', 'organizationalPerson', 'inetOrgPerson', 'posixAccount', 'shadowAccount']

        # Attributes for a user entry
        attributes = {'cn' : user, 'sn' : user, 'givenName' : user, 'uid' : user, 'uidNumber' : self.lastuid, 'gidNumber' : self.lastgid, 'homeDirectory' : f'/home/{user}', 'loginShell' : '/usr/bin/git-shell', 'gecos' : 'SystemUser', 'shadowLastChange' : self.lastpwchangenow(), 'shadowMax' : '45', 'userPassword' : password, 'mail' : f'{user}@{self.domain}' }

        # Return boolean value of new user entry
        rcode1 = self.ldapconnection.add(f'uid={user},{self.base}', objectClass, attributes)


        # Add new user to log
        attributes['description'] = 'ADD'
        attributes['uid'] = str(lastlog)

        if rcode1:
            rcode2 = self.ldapconnection.add(f'uid={str(lastlog)},{self.logbase}', objectClass, attributes)
        else:
            return False

        if rcode2:
            self.setlastlog(lastlog)
            self.lastlog=lastlog

        # Return True only if both entries was successful
        return rcode1 and rcode2

    def changepassword(self, user, newpass):
        '''
            Change password of user to newpass

            user := string containing username
            newpass := string containing new password
        '''
        # This variable holds boolean indicating successful change of user password
        chpassbool = False

        # This variable holds boolean indicating successful change of shadowLastChange value to current time
        chlastchangebool = False

        USERATTRIBUTES=['cn' , 'sn', 'givenName', 'uid', 'uidNumber' , 'gidNumber', 'homeDirectory', 'loginShell', 'gecos' , 'shadowLastChange', 'shadowMax', 'userPassword', 'mail']

        OBJECTCLASSES = ['top', 'person', 'organizationalPerson', 'inetOrgPerson', 'posixAccount', 'shadowAccount']

        self.ldapconnection.search(search_base=f'uid={user},{self.base}',search_filter='(objectClass=person)', attributes=USERATTRIBUTES)

        userdata = self.ldapconnection.response[0]
        userdata['attributes']['description'] = 'CHANGEPASS'

        lastlog = self.lastlog + 1
        userdata['attributes']['uid'] = str(lastlog)


        chpassbool = self.ldapconnection.modify(f'uid={user},{self.base}', {'userPassword': (MODIFY_REPLACE,[newpass])})
        chlastchangebool = self.ldapconnection.modify(f'uid={user},{self.base}', {'shadowLastChange' : (MODIFY_REPLACE,[self.lastpwchangenow()])})

        if chpassbool and chlastchangebool:
            rcode1 = self.ldapconnection.add(f'uid={lastlog},{self.logbase}', OBJECTCLASSES, userdata['attributes'])

            if rcode1:
                self.setlastlog(lastlog)
                self.lastlog = lastlog
                return True
            else:
                return False
        else:
            return False

    def delete(self, user):
        '''
           Delete user given username of user

           user := string containing username
        '''

        USERATTRIBUTES=['cn' , 'sn', 'givenName', 'uid', 'uidNumber' , 'gidNumber', 'homeDirectory', 'loginShell', 'gecos' , 'shadowLastChange', 'shadowMax', 'userPassword', 'mail']

        OBJECTCLASSES = ['top', 'person', 'organizationalPerson', 'inetOrgPerson', 'posixAccount', 'shadowAccount']

        self.ldapconnection.search(search_base=f'uid={user},{self.base}',search_filter='(objectClass=person)', attributes=USERATTRIBUTES)

        userdata = self.ldapconnection.response[0]
        userdata['attributes']['description'] = 'DELETE'

        lastlog = self.lastlog + 1
        userdata['attributes']['uid'] = str(lastlog)



        rcode1 = self.ldapconnection.delete(f'uid={user},{self.base}')

        if rcode1:
            rcode2 = self.ldapconnection.add(f'uid={lastlog},{self.logbase}', OBJECTCLASSES, userdata['attributes'])

            if rcode2:
                self.setlastlog(lastlog)
                self.lastlog = lastlog
                return True
            else:
                return False
        else:
            return False

    def getpassword(self, user):
        '''
           Retrive password of a user 

           user := string containing username
        '''

        # Search LDAP entries that have object class inetOrgPerson and uid attribute equal to given user field
        self.ldapconnection.search(search_base=self.base,search_filter=f'(&(objectClass=inetOrgPerson)(uid={user}))', attributes=['userPassword'])

        ## Check if user exists
        if self.ldapconnection.response == []:
            return False;

        # Return userPassword attribute from the response
        userpass = self.ldapconnection.response[0]['attributes']['userPassword'][0]
        
        if type(userpass) is bytes:
            userpass = userpass.decode('utf-8')

        return userpass