49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
from typing import Union, List
|
|
|
|
import ldap3
|
|
from zope import component
|
|
from zope.interface import implementer
|
|
|
|
from .UWLDAPRecord import UWLDAPRecord
|
|
from .utils import dn_to_uid
|
|
from ceo_common.interfaces import IUWLDAPService, IConfig
|
|
|
|
|
|
@implementer(IUWLDAPService)
|
|
class UWLDAPService:
|
|
def __init__(self):
|
|
cfg = component.getUtility(IConfig)
|
|
self.uwldap_server_url = cfg.get('uwldap_server_url')
|
|
self.uwldap_base = cfg.get('uwldap_base')
|
|
|
|
def _get_conn(self) -> ldap3.Connection:
|
|
return ldap3.Connection(
|
|
self.uwldap_server_url, auto_bind=True, read_only=True,
|
|
raise_exceptions=True)
|
|
|
|
def get_user(self, username: str) -> Union[UWLDAPRecord, None]:
|
|
conn = self._get_conn()
|
|
conn.search(
|
|
self.uwldap_base, f'(uid={username})',
|
|
attributes=UWLDAPRecord.ldap_attributes, size_limit=1)
|
|
if not conn.entries:
|
|
return None
|
|
return UWLDAPRecord.deserialize_from_ldap(conn.entries[0])
|
|
|
|
def get_programs_for_users(self, usernames: List[str]) -> List[Union[str, None]]:
|
|
filter_str = '(|' + ''.join([f'(uid={uid})' for uid in usernames]) + ')'
|
|
programs = [None] * len(usernames)
|
|
user_indices = {uid: i for i, uid in enumerate(usernames)}
|
|
|
|
conn = self._get_conn()
|
|
conn.search(
|
|
self.uwldap_base, filter_str, attributes=['ou'],
|
|
size_limit=len(usernames))
|
|
for entry in conn.entries:
|
|
uid = dn_to_uid(entry.entry_dn)
|
|
idx = user_indices[uid]
|
|
program = entry.ou.value
|
|
if program:
|
|
programs[idx] = program
|
|
return programs
|