import copy from typing import List, Dict, Union from zope import component from zope.interface import implementer from .utils import strings_to_bytes, bytes_to_strings, dn_to_uid from ceo_common.interfaces import ILDAPService, IGroup, IConfig, IUser @implementer(IGroup) class Group: def __init__( self, cn: str, gid_number: int, members: Union[List[str], None] = None, ): self.cn = cn self.gid_number = gid_number # this is a list of the DNs of the members in this group self.member_DNs = members or [] cfg = component.getUtility(IConfig) self.dn = f'cn={cn},{cfg.get("ldap_groups_base")}' self.ldap_users_base = cfg.get('ldap_users_base') self.ldap_srv = component.getUtility(ILDAPService) def __repr__(self) -> str: lines = [ 'dn: ' + self.dn, 'cn: ' + self.cn, 'gidNumber: ' + str(self.gid_number), 'objectClass: top', 'objectClass: group', 'objectClass: posixGroup', ] for member_dn in self.member_DNs: lines.append('uniqueMember: ' + member_dn) return '\n'.join(lines) def add_to_ldap(self): self.ldap_srv.add_group(self) def serialize_for_modlist(self) -> Dict[str, List[bytes]]: data = { 'cn': [self.cn], 'gidNumber': [str(self.gid_number)], 'objectClass': [ 'top', 'group', 'posixGroup', ], } if self.member_DNs: data['uniqueMember'] = self.member_DNs return strings_to_bytes(data) @staticmethod def deserialize_from_dict(data: Dict[str, List[bytes]]) -> IGroup: data = bytes_to_strings(data) return Group( cn=data['cn'][0], gid_number=int(data['gidNumber'][0]), members=data.get('uniqueMember'), ) def add_member(self, username: str): new_group = copy.copy(self) new_group.member_DNs = self.member_DNs.copy() new_group.member_DNs.append(f'uid={username},{self.ldap_users_base}') self.ldap_srv.modify_group(self, new_group) self.member_DNs = new_group.member_DNs def remove_member(self, username: str): new_group = copy.copy(self) new_group.member_DNs = self.member_DNs.copy() new_group.member_DNs.remove(f'uid={username},{self.ldap_users_base}') self.ldap_srv.modify_group(self, new_group) self.member_DNs = new_group.member_DNs def get_members(self) -> List[IUser]: members = [] for dn in self.member_DNs: members.append(self.ldap_srv.get_user(dn_to_uid(dn))) return members