pyceo/ceod/model/LDAPService.py

import copy
import grp
import pwd

import ldap
import ldap.modlist
from zope import component
from zope.interface import implementer

from ceo_common.errors import UserNotFoundError, GroupNotFoundError
from ceo_common.interfaces import ILDAPService, IKerberosService, IConfig, IUser, IGroup
from .User import User
from .Group import Group
from .SudoRole import SudoRole


@implementer(ILDAPService)
class LDAPService:
    def __init__(self):
        cfg = component.getUtility(IConfig)
        self.ldap_admin_principal = cfg.get('ldap_admin_principal')
        self.ldap_server_url = cfg.get('ldap_server_url')
        self.ldap_users_base = cfg.get('ldap_users_base')
        self.ldap_groups_base = cfg.get('ldap_groups_base')
        self.member_min_id = cfg.get('member_min_id')
        self.member_max_id = cfg.get('member_max_id')
        self.club_min_id = cfg.get('club_min_id')
        self.club_max_id = cfg.get('club_max_id')

    def _get_ldap_conn(self, gssapi_bind: bool = True) -> ldap.ldapobject.LDAPObject:
        # TODO: cache the connection
        conn = ldap.initialize(self.ldap_server_url)
        if gssapi_bind:
            self._gssapi_bind(conn)
        return conn

    def _gssapi_bind(self, conn: ldap.ldapobject.LDAPObject):
        krb_srv = component.getUtility(IKerberosService)
        for i in range(2):
            try:
                conn.sasl_gssapi_bind_s()
                return
            except ldap.LOCAL_ERROR as err:
                if 'Ticket expired' in err.args[0]['info']:
                    krb_srv.kinit()
                    continue
                raise err
        raise Exception('could not perform GSSAPI bind')

    def get_user(self, username: str) -> IUser:
        conn = self._get_ldap_conn(False)
        base = f'uid={username},{self.ldap_users_base}'
        try:
            _, result = conn.search_s(base, ldap.SCOPE_BASE)[0]
            return User.deserialize_from_ldap(result)
        except ldap.NO_SUCH_OBJECT:
            raise UserNotFoundError()

    def get_group(self, cn: str) -> IGroup:
        conn = self._get_ldap_conn(False)
        base = f'cn={cn},{self.ldap_groups_base}'
        try:
            _, result = conn.search_s(base, ldap.SCOPE_BASE)[0]
            return Group.deserialize_from_ldap(result)
        except ldap.NO_SUCH_OBJECT:
            raise GroupNotFoundError()

    def _get_next_uid(self, conn: ldap.ldapobject.LDAPObject, min_id: int, max_id: int) -> int:
        """Gets the next available UID number between min_id and max_id, inclusive."""
        def uid_exists(uid: int) -> bool:
            try:
                pwd.getpwuid(uid)
                return True
            except KeyError:
                return False

        def gid_exists(gid: int) -> bool:
            try:
                grp.getgrgid(gid)
                return True
            except KeyError:
                return False

        def ldap_uid_or_gid_exists(uid: int) -> bool:
            results = conn.search_s(
                self.ldap_users_base, ldap.SCOPE_ONELEVEL,
                f'(|(uidNumber={uid})(gidNumber={uid}))')
            return len(results) > 0

        # TODO: replace this with binary search
        for uid in range(min_id, max_id + 1):
            if uid_exists(uid) or gid_exists(uid) or ldap_uid_or_gid_exists(uid):
                continue
            return uid
        raise Exception('no UIDs remaining')

    def add_sudo_role(self, uid: str):
        conn = self._get_ldap_conn()
        sudo_role = SudoRole(uid)
        modlist = ldap.modlist.addModlist(sudo_role.serialize_for_ldap())
        conn.add_s(sudo_role.dn, modlist)

    def remove_sudo_role(self, uid: str):
        conn = self._get_ldap_conn()
        sudo_role = SudoRole(uid)
        conn.delete_s(sudo_role.dn)

    def add_user(self, user: IUser) -> IUser:
        if user.is_club():
            min_id, max_id = self.club_min_id, self.club_max_id
        else:
            min_id, max_id = self.member_min_id, self.member_max_id
        conn = self._get_ldap_conn()
        uid_number = self._get_next_uid(conn, min_id, max_id)
        new_user = copy.copy(user)
        new_user.uid_number = uid_number
        new_user.gid_number = uid_number

        modlist = ldap.modlist.addModlist(new_user.serialize_for_ldap())
        conn.add_s(new_user.dn, modlist)

        return new_user

    def modify_user(self, old_user: IUser, new_user: IUser):
        conn = self._get_ldap_conn()
        modlist = ldap.modlist.modifyModlist(
            old_user.serialize_for_ldap(),
            new_user.serialize_for_ldap(),
        )
        conn.modify_s(old_user.dn, modlist)

    def remove_user(self, user: IUser):
        conn = self._get_ldap_conn()
        conn.delete_s(user.dn)

    def add_group(self, group: IGroup) -> IGroup:
        conn = self._get_ldap_conn()
        # make sure that the caller initialized the GID number
        assert group.gid_number
        modlist = ldap.modlist.addModlist(group.serialize_for_ldap())
        conn.add_s(group.dn, modlist)
        return group

    def modify_group(self, old_group: IGroup, new_group: IGroup):
        conn = self._get_ldap_conn()
        modlist = ldap.modlist.modifyModlist(
            old_group.serialize_for_ldap(),
            new_group.serialize_for_ldap(),
        )
        conn.modify_s(old_group.dn, modlist)

    def remove_group(self, group: IGroup):
        conn = self._get_ldap_conn()
        conn.delete_s(group.dn)