synchronize-dolibarr-to-ldap/main.py

#!/usr/bin/env python3

from dolibarrpy import Dolibarrpy
from flask import Flask, abort, request
from ldap3 import ALL, Connection, ObjectDef, Reader, Server, WritableEntry, Writer

import config


def main():
    dolibarr_client = Dolibarrpy(url=config.DOLIBARR_API_BASE, token=config.DOLIBARR_API_TOKEN, timeout=16, debug=config.DOLIBARR_API_DEBUG)

    ldap_server = Server(config.LDAP_HOST, config.LDAP_PORT, get_info=ALL)
    with Connection(ldap_server, config.LDAP_BIND_USER, config.LDAP_BIND_PASSWORD) as ldap_conn:
        if config.LDAP_USERS_EXTRA_FIELDS or config.LDAP_USERS_EXTRA_OBJECT_CLASSES:
            manage_users_extra_fields(ldap_conn, dolibarr_client)

        if config.LDAP_GROUPS_EXTRA_FIELDS or config.LDAP_GROUPS_EXTRA_OBJECT_CLASSES:
            manage_groups_extra_fields(ldap_conn, dolibarr_client)


def manage_users_extra_fields(ldap_conn: Connection, dolibarr_client: Dolibarrpy):
    dolibarr_users = dolibarr_client.find_all_users()
    for dolibarr_user in dolibarr_users:
        manage_user_extra_fields(ldap_conn, dolibarr_user, dolibarr_client)


def manage_user_extra_fields(ldap_conn: Connection, dolibarr_user: dict, dolibarr_client: Dolibarrpy):
    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_writer = Writer.from_cursor(users_reader, object_def=obj_user)
    if users_writer.entries:
        ldap_user = users_writer[0]
    else:
        attrs = {
            'cn': f"{dolibarr_user['firstname']} {dolibarr_user['lastname']}".strip(),
            'givenName': dolibarr_user['firstname'],
            'sn': dolibarr_user['lastname'],
            'mail': dolibarr_user['email'],
            'street': dolibarr_user['address'],
            'postalCode': dolibarr_user['zip'],
            'l': dolibarr_user['town'],
            'mobile': dolibarr_user['user_mobile'],
            'uidNumber': dolibarr_user['id'],
            'gidNumber': dolibarr_user['id'],
            'homeDirectory': f"/home/{login}",
        }
        for key, value in list(attrs.items()):
            if not value:
                del attrs[key]
        ldap_conn.add(f"uid={login},{config.LDAP_USERS_OU}", ["top", "inetOrgPerson", "posixAccount", "shadowAccount"], attrs)
        users_reader.search()
        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)
    append_extra_group_fields_to_ldap_user(ldap_user, dolibarr_user, dolibarr_client)
    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_attrs, ldap_attr = extra_field.split(':')
        dolibarr_attrs = dolibarr_attrs.split('|')
        values = []
        for dolibarr_attr in dolibarr_attrs:
            if dolibarr_attr.startswith('GROUP__'):
                continue
            if dolibarr_attr.endswith('[]'):
                dolibarr_attr = dolibarr_attr[:-2]
                value = dolibarr_user['array_options'][f'options_{dolibarr_attr}']
                value = value.split() if value else []
                values += value
            else:
                value = dolibarr_user['array_options'][f'options_{dolibarr_attr}']
                if value:
                    values.append(value)
        if not values:
            continue
        if ldap_attr.endswith('[]'):
            ldap_attr = ldap_attr[:-2]
            value = values
        else:
            value = values[0]
        setattr(ldap_user, ldap_attr, value)


def append_extra_group_fields_to_ldap_user(ldap_user: WritableEntry, dolibarr_user: dict, dolibarr_client: Dolibarrpy):
    if not any(dolibarr_attr.startswith('GROUP')
               for extra_field in config.LDAP_USERS_EXTRA_FIELDS
               for dolibarr_attr in extra_field.split(':')[0].split('|')):
        return

    user_id = dolibarr_user['id']
    dolibarr_groups: list[dict] = dolibarr_client.get_user_groups_uid(user_id)

    for extra_field in config.LDAP_USERS_EXTRA_FIELDS:
        dolibarr_attrs, ldap_attr = extra_field.split(':')
        dolibarr_attrs = dolibarr_attrs.split('|')
        values = []
        for dolibarr_attr in dolibarr_attrs:
            if not dolibarr_attr.startswith('GROUP__'):
                continue
            dolibarr_attr = dolibarr_attr[7:]
            for dolibarr_group in dolibarr_groups:
                if dolibarr_attr.endswith('[]'):
                    dolibarr_attr = dolibarr_attr[:-2]
                    value = dolibarr_group.get(dolibarr_attr, dolibarr_group['array_options'][f'options_{dolibarr_attr}'])
                    value = value.split() if value else []
                    values += value
                else:
                    value = dolibarr_group.get(dolibarr_attr, dolibarr_group['array_options'][f'options_{dolibarr_attr}'])
                    if value:
                        values.append(value)
        if not values:
            continue
        if ldap_attr.endswith('[]'):
            ldap_attr = ldap_attr[:-2]
            value = values
        else:
            value = values[0]
        setattr(ldap_user, ldap_attr, value)
        

def manage_groups_extra_fields(ldap_conn: Connection, dolibarr_client: Dolibarrpy):
    dolibarr_groups = dolibarr_client.call_list_api('users/groups')
    for dolibarr_group in dolibarr_groups:
        manage_group_extra_fields(ldap_conn, dolibarr_group)


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_writer = Writer.from_cursor(groups_reader, object_def=obj_group)
    if groups_writer.entries:
        ldap_group = groups_writer[0]
    else:
        attrs = {
            'cn': name,
            'gidNumber': dolibarr_group['id'],
        }
        for key, value in list(attrs.items()):
            if not value:
                del attrs[key]
        ldap_conn.add(f"cn={name},{config.LDAP_GROUPS_OU}", ["top", "posixGroup"], attrs)
        groups_reader.search()
        groups_writer = Writer.from_cursor(groups_reader, object_def=obj_group)
        ldap_group = groups_writer[0]
    append_extra_fields_to_ldap_group(ldap_group, dolibarr_group)
    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_attrs, ldap_attr = extra_field.split(':')
        dolibarr_attrs = dolibarr_attrs.split('|')
        values = []
        for dolibarr_attr in dolibarr_attrs:
            if dolibarr_attr.endswith('[]'):
                dolibarr_attr = dolibarr_attr[:-2]
                value = dolibarr_group['array_options'][f'options_{dolibarr_attr}']
                value = value.split() if value else []
                values += value
            else:
                value = dolibarr_group['array_options'][f'options_{dolibarr_attr}']
                if value:
                    values.append(value)
        if not values:
            continue
        if ldap_attr.endswith('[]'):
            ldap_attr = ldap_attr[:-2]
            value = values
        else:
            value = values[0]
        setattr(ldap_group, ldap_attr, value)


flask_app = Flask(__name__)

@flask_app.post('/webhook')
def webhook_receiver():
    data = request.json
    if 'triggercode' not in data or 'object' not in data:
        abort(400)
    triggercode = data['triggercode']
    obj = data['object']
    dolibarr_client = Dolibarrpy(url=config.DOLIBARR_API_BASE, token=config.DOLIBARR_API_TOKEN, timeout=16, debug=config.DOLIBARR_API_DEBUG)
    ldap_server = Server(config.LDAP_HOST, config.LDAP_PORT, get_info=ALL)
    if triggercode.startswith('USER_'):
        with Connection(ldap_server, config.LDAP_BIND_USER, config.LDAP_BIND_PASSWORD) as ldap_conn:
            manage_user_extra_fields(ldap_conn, obj)
    elif triggercode.startswith('USERGROUP_'):
        with Connection(ldap_server, config.LDAP_BIND_USER, config.LDAP_BIND_PASSWORD) as ldap_conn:
            manage_group_extra_fields(ldap_conn, obj)
    else:
        abort(400)
    return "", 204


if __name__ == '__main__':
    main()