import json from typing import List, Union import ldap3 from zope import component from zope.interface import implementer from .utils import dn_to_uid from ceo_common.errors import UserAlreadyInGroupError, UserNotInGroupError, \ UserNotFoundError from ceo_common.interfaces import ILDAPService, IGroup from ceo_common.logger_factory import logger_factory logger = logger_factory(__name__) @implementer(IGroup) class Group: def __init__( self, cn: str, gid_number: int, description: Union[str, None] = None, members: Union[List[str], None] = None, ldap3_entry: Union[ldap3.Entry, None] = None, user_cn: Union[str, None] = None, ): self.cn = cn self.gid_number = gid_number self.description = description # this is a list of the usernames of the members in this group self.members = members or [] self.ldap3_entry = ldap3_entry self.user_cn = user_cn self.ldap_srv = component.getUtility(ILDAPService) def __repr__(self) -> str: return json.dumps(self.to_dict(), indent=2) def to_dict(self): data = { 'cn': self.cn, 'gid_number': self.gid_number, } description = None if self.description: description = self.description elif self.user_cn: # for clubs, the human-readable description is stored in the # 'cn' attribute of the associated user description = self.user_cn else: try: # TODO: only fetch the CN to save bandwidth user = self.ldap_srv.get_user(self.cn) description = user.cn self.user_cn = user.cn except UserNotFoundError: # some groups, like syscom, don't have an associated user pass if description: data['description'] = description # to_dict() is usually called for display purposes, so get some more # information to display data['members'] = self.ldap_srv.get_display_info_for_users(self.members) return data def add_to_ldap(self): self.ldap_srv.add_group(self) def remove_from_ldap(self): self.ldap_srv.remove_group(self) @staticmethod def deserialize_from_ldap(entry: ldap3.Entry) -> IGroup: """Deserialize this group from an LDAP entry.""" attrs = entry.entry_attributes_as_dict return Group( cn=attrs['cn'][0], gid_number=attrs['gidNumber'][0], description=attrs.get('description', [None])[0], members=[ dn_to_uid(dn) for dn in attrs.get('uniqueMember', []) ], ldap3_entry=entry, ) def add_member(self, username: str): dn = self.ldap_srv.uid_to_dn(username) try: with self.ldap_srv.entry_ctx_for_group(self) as entry: entry.uniqueMember.add(dn) except ldap3.core.exceptions.LDAPAttributeOrValueExistsResult as err: logger.warning(err) raise UserAlreadyInGroupError() self.members.append(username) def remove_member(self, username: str): dn = self.ldap_srv.uid_to_dn(username) try: with self.ldap_srv.entry_ctx_for_group(self) as entry: entry.uniqueMember.delete(dn) except ldap3.core.exceptions.LDAPCursorError as err: logger.warning(err) raise UserNotInGroupError() self.members.remove(username) def set_members(self, usernames: List[str]): DNs = [ self.ldap_srv.uid_to_dn(username) for username in usernames ] with self.ldap_srv.entry_ctx_for_group(self) as entry: entry.uniqueMember = DNs self.members = usernames