import logging from ldap3 import Server, Connection, ALL from ldap3.core.exceptions import LDAPException class LLDAP: def __init__( self, ldap_server: str, bind_dn: str, bind_password: str, user_dn: str = "ou=people,dc=example,dc=com", group_dn: str = "ou=people,dc=example,dc=com" ): self.bind_dn = bind_dn self.bind_password = bind_password self.server = Server(ldap_server, get_info=ALL) self.conn = None self.user_dn = user_dn self.group_dn = group_dn def do_bind(self): try: if not self.conn: self.conn = Connection( self.server, user=self.bind_dn, password=self.bind_password, auto_bind=True ) logging.info("Successfully bound to LDAP server.") except LDAPException as e: logging.error(f"Failed to bind to LDAP server: {e}") raise def do_unbind(self): if self.conn: self.conn.unbind() self.conn = None logging.info("Successfully unbound from the LDAP server.") def list_groups(self): try: self.do_bind() self.conn.search(self.group_dn, "(objectClass=*)", attributes=['cn']) entries = self.conn.entries return entries except LDAPException as e: logging.error(f"Error during LDAP search: {e}") raise finally: self.do_unbind() def list_users(self): try: self.do_bind() self.conn.search(self.user_dn, "(uid=*)", attributes=['dn']) entries = self.conn.entries return entries except LDAPException as e: logging.error(f"Error during LDAP search: {e}") raise finally: self.do_unbind()