sheppy/atlantis-event-dispatcher
1
0
Fork 0
mirror of https://github.com/FAUSheppy/atlantis-event-dispatcher synced 2025-12-09 07:48:33 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: ldap group based selection

Browse Source
This commit is contained in:
Johnathen Sheppard
2023-07-16 13:21:29 +02:00
parent a81fafb9c7
commit 749282cb92
1 changed files with 20 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
ldaptools.py
View File

@@ -16,6 +16,12 @@ class Person:
self.email = email self.email = email
self.phone = phone self.phone = phone
def __eq__(self, other):
return other.cn == self.cn
def __hash__(self):
return hash(self.cn)
def ldap_query(search_filter, ldap_args, alt_base_dn=None): def ldap_query(search_filter, ldap_args, alt_base_dn=None):
ldap_server = ldap_args["LDAP_SERVER"] ldap_server = ldap_args["LDAP_SERVER"]
@@ -48,11 +54,14 @@ def _person_from_search_result(cn, entry):
return Person(cn, username, name, email, phone) return Person(cn, username, name, email, phone)
def get_user_by_uid(username, ldap_args): def get_user_by_uid(username, ldap_args, uid_is_cn=False):
if not username: if not username:
return None return None
if uid_is_cn:
username = username.split(",")[0].split("=")[1]
search_filter = "(&(objectClass=inetOrgPerson)(uid={username}))".format(username=username) search_filter = "(&(objectClass=inetOrgPerson)(uid={username}))".format(username=username)
results = ldap_query(search_filter, ldap_args) results = ldap_query(search_filter, ldap_args)
@@ -68,8 +77,11 @@ def get_members_of_group(group, ldap_args):
if not group: if not group:
return [] return []
search_filter = "(&(objectClass=groupOfNames)(cn={group_name})".format(group) search_filter = "(&(objectClass=groupOfNames)(cn={group_name}))".format(group_name=group)
results = ldap_query(search_filter, ldap_args)
# TODO wtf is this btw??
groups_dn = ",".join([ s.replace("People","groups") for s in base_dn.split(",")])
results = ldap_query(search_filter, ldap_args, alt_base_dn=groups_dn)
if not results: if not results:
return [] return []
@@ -80,15 +92,12 @@ def get_members_of_group(group, ldap_args):
persons = [] persons = []
for member in members: for member in members:
user_dn = member.decode("utf-8") user_cn = member.decode("utf-8")
user_filter = "(objectClass=inetOrgPerson)" person_obj = get_user_by_uid(user_cn, ldap_args, uid_is_cn=True)
results = ldap_query(user_filter, ldap_args, alt_base_dn=user_dn)
if not results: if not person_obj:
continue continue
cn, entry = results[0]
person_obj = _person_from_search_result(cn, entry)
persons.append(person_obj) persons.append(person_obj)
return persons return persons
@@ -106,6 +115,6 @@ def select_targets(users, groups, ldap_args, admin_group="pki"):
persons += get_members_of_group(group, ldap_args) persons += get_members_of_group(group, ldap_args)
else: else:
# send to administrators # # send to administrators #
persons += get_members_of_group() persons += get_members_of_group(admin_group, ldap_args)
return persons return set(persons)