ynerant/synchronize-dolibarr-to-ldap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Synchronization is now operated from dolibarr data

Browse Source
This commit is contained in:
Emmy D'Anello 2025-02-17 12:16:28 +01:00
parent 01fdd6cc01
commit ed25f0ba2b
Signed by: ynerant
GPG Key ID: 3A75C55819C8CF85
1 changed files with 91 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

146
main.py Normal file → Executable file
View File

@ -1,7 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
from dolibarrpy import Dolibarrpy from dolibarrpy import Dolibarrpy
from ldap3 import ALL, Connection, ObjectDef, Reader, Server, Writer from ldap3 import ALL, Connection, ObjectDef, Reader, Server, WritableEntry, Writer
import config import config
@ -20,71 +20,107 @@ def main():
def manage_users_extra_fields(ldap_conn: Connection, dolibarr_client: Dolibarrpy): def manage_users_extra_fields(ldap_conn: Connection, dolibarr_client: Dolibarrpy):
dolibarr_users = dolibarr_client.find_all_users() dolibarr_users = dolibarr_client.find_all_users()
for dolibarr_user in dolibarr_users:
manage_user_extra_fields(ldap_conn, dolibarr_user)
obj_inetorgperson = ObjectDef(['inetOrgPerson'] + config.LDAP_USERS_EXTRA_OBJECT_CLASSES, ldap_conn)
users_reader = Reader(ldap_conn, obj_inetorgperson, config.LDAP_USERS_OU) def manage_user_extra_fields(ldap_conn: Connection, dolibarr_user: dict):
login = dolibarr_user['login']
obj_inetorgperson = ObjectDef(['top', 'inetOrgPerson', 'posixAccount'], ldap_conn)
obj_user = ObjectDef(['top', 'inetOrgPerson', 'posixAccount'] + config.LDAP_GROUPS_EXTRA_OBJECT_CLASSES, ldap_conn)
users_reader = Reader(ldap_conn, obj_inetorgperson, config.LDAP_USERS_OU, f"uid:={login}")
users_reader.search() users_reader.search()
users_writer = Writer.from_cursor(users_reader) users_writer = Writer.from_cursor(users_reader, object_def=obj_user)
for ldap_user in users_writer: if users_writer.entries:
uid = ldap_user.uid ldap_user = users_writer[0]
for dolibarr_user in dolibarr_users: else:
if dolibarr_user['login'] == uid: attrs = {
break 'cn': f"{dolibarr_user['firstname']} {dolibarr_user['lastname']}".strip(),
else: 'givenName': dolibarr_user['firstname'],
continue 'sn': dolibarr_user['lastname'],
'mail': dolibarr_user['email'],
for extra_object_class in config.LDAP_USERS_EXTRA_OBJECT_CLASSES: 'street': dolibarr_user['address'],
if extra_object_class not in ldap_user.objectClass: 'postalCode': dolibarr_user['zip'],
ldap_user.objectClass.append(extra_object_class) 'l': dolibarr_user['town'],
'mobile': dolibarr_user['user_mobile'],
for extra_field in config.LDAP_USERS_EXTRA_FIELDS: 'uidNumber': dolibarr_user['id'],
dolibarr_attr, ldap_attr = extra_field.split(':') 'gidNumber': dolibarr_user['id'],
if dolibarr_attr.endswith('[]'): 'homeDirectory': f"/home/{login}",
dolibarr_attr = dolibarr_attr[:-2] }
value = dolibarr_user['array_options'][f'options_{dolibarr_attr}'] for key, value in list(attrs.items()):
value = value.split() if value else [] if not value:
setattr(ldap_user, ldap_attr, value) del attrs[key]
else: ldap_conn.add(f"uid={login},{config.LDAP_USERS_OU}", ["top", "inetOrgPerson", "posixAccount", "shadowAccount"], attrs)
value = dolibarr_user['array_options'][f'options_{dolibarr_attr}'] or "" users_reader.search()
setattr(ldap_user, ldap_attr, value) users_writer = Writer.from_cursor(users_reader, object_def=obj_user)
ldap_user = users_writer[0]
append_extra_fields_to_ldap_user(ldap_user, dolibarr_user)
users_writer.commit() users_writer.commit()
def append_extra_fields_to_ldap_user(ldap_user: WritableEntry, dolibarr_user: dict):
for extra_object_class in config.LDAP_USERS_EXTRA_OBJECT_CLASSES:
if extra_object_class not in ldap_user.objectClass:
ldap_user.objectClass += extra_object_class
for extra_field in config.LDAP_USERS_EXTRA_FIELDS:
dolibarr_attr, ldap_attr = extra_field.split(':')
if dolibarr_attr.endswith('[]'):
dolibarr_attr = dolibarr_attr[:-2]
value = dolibarr_user['array_options'][f'options_{dolibarr_attr}']
value = value.split() if value else []
setattr(ldap_user, ldap_attr, value)
else:
value = dolibarr_user['array_options'][f'options_{dolibarr_attr}'] or ""
setattr(ldap_user, ldap_attr, value)
def manage_groups_extra_fields(ldap_conn: Connection, dolibarr_client: Dolibarrpy): def manage_groups_extra_fields(ldap_conn: Connection, dolibarr_client: Dolibarrpy):
dolibarr_groups = dolibarr_client.call_list_api('users/groups') dolibarr_groups = dolibarr_client.call_list_api('users/groups')
for dolibarr_group in dolibarr_groups:
manage_group_extra_fields(ldap_conn, dolibarr_group)
obj_posixgroup = ObjectDef(['posixGroup'] + config.LDAP_GROUPS_EXTRA_OBJECT_CLASSES, ldap_conn)
groups_reader = Reader(ldap_conn, obj_posixgroup, config.LDAP_GROUPS_OU) def manage_group_extra_fields(ldap_conn: Connection, dolibarr_group: dict):
name = dolibarr_group['name']
obj_posixgroup = ObjectDef(['posixGroup'], ldap_conn)
obj_group = ObjectDef(['posixGroup'] + config.LDAP_GROUPS_EXTRA_OBJECT_CLASSES, ldap_conn)
groups_reader = Reader(ldap_conn, obj_posixgroup, config.LDAP_GROUPS_OU, f"cn:={name}")
groups_reader.search() groups_reader.search()
groups_writer = Writer.from_cursor(groups_reader) groups_writer = Writer.from_cursor(groups_reader, object_def=obj_group)
for ldap_group in groups_writer: if groups_writer.entries:
print(ldap_group) ldap_group = groups_writer[0]
name = ldap_group.cn else:
for dolibarr_group in dolibarr_groups: attrs = {
if dolibarr_group['nom'] == name: 'cn': name,
break 'gidNumber': dolibarr_group['id'],
else: }
continue for key, value in list(attrs.items()):
if not value:
for extra_object_class in config.LDAP_GROUPS_EXTRA_OBJECT_CLASSES: del attrs[key]
if extra_object_class not in ldap_group.objectClass: ldap_conn.add(f"cn={name},{config.LDAP_GROUPS_OU}", ["top", "posixGroup"], attrs)
ldap_group.objectClass.append(extra_object_class) groups_reader.search()
groups_writer = Writer.from_cursor(groups_reader, object_def=obj_group)
for extra_field in config.LDAP_GROUPS_EXTRA_FIELDS: ldap_group = groups_writer[0]
dolibarr_attr, ldap_attr = extra_field.split(':') append_extra_fields_to_ldap_group(ldap_group, dolibarr_group)
if dolibarr_attr.endswith('[]'):
dolibarr_attr = dolibarr_attr[:-2]
print(dolibarr_group)
value = dolibarr_group['array_options'][f'options_{dolibarr_attr}']
value = value.split() if value else []
print(ldap_attr, value)
setattr(ldap_group, ldap_attr, value)
else:
value = dolibarr_group['array_options'][f'options_{dolibarr_attr}'] or ""
print(ldap_attr, value)
setattr(ldap_group, ldap_attr, value)
groups_writer.commit() groups_writer.commit()
def append_extra_fields_to_ldap_group(ldap_group: WritableEntry, dolibarr_group: dict):
for extra_object_class in config.LDAP_GROUPS_EXTRA_OBJECT_CLASSES:
if extra_object_class not in ldap_group.objectClass:
ldap_group.objectClass += extra_object_class
for extra_field in config.LDAP_GROUPS_EXTRA_FIELDS:
dolibarr_attr, ldap_attr = extra_field.split(':')
if dolibarr_attr.endswith('[]'):
dolibarr_attr = dolibarr_attr[:-2]
value = dolibarr_group['array_options'][f'options_{dolibarr_attr}']
value = value.split() if value else []
setattr(ldap_group, ldap_attr, value)
else:
value = dolibarr_group['array_options'][f'options_{dolibarr_attr}'] or ""
setattr(ldap_group, ldap_attr, value)
if __name__ == '__main__': if __name__ == '__main__':
main() main()