Python-Projekte/ldap/convert_ldap_to_db.py

import json
from pathlib import Path

base_dir = Path(__file__).parent / "Becker-Tiemann"


def read_ldap_json(json_file: str) -> dict:
    with open(base_dir / json_file, "r", encoding="latin-1") as frh:
        groups_temp = json.load(frh)

    res = {}
    for entry in groups_temp["entries"]:
        res[entry["dn"]] = entry["attributes"]

    return res


def read_or_create_selected_groups(json_file: str, groups: dict) -> dict:
    sel_groups = []
    if Path(base_dir / json_file).exists():
        with open(base_dir / json_file, "r", encoding="latin-1") as frh:
            sel_groups = json.load(frh)
    else:
        sel_groups = [dn for dn, attrs in groups.items() if len(attrs.get("member", [])) > 0]
        with open(base_dir / json_file, "w", encoding="latin-1") as fwh:
            json.dump(sel_groups, fwh, indent=2)
    return sel_groups


def print_group_membership(groups: dict, sel_groups: list):
    for dn, attrs in groups.items():
        group_members = [member_dn for member_dn in attrs.get("member", []) if member_dn in groups]
        if group_members:
            print("\n" + dn, len(group_members))
            for member_dn in group_members:
                print(" * " + member_dn)


def get_subgroups(groups: dict, sel_groups: list):
    res = []
    for group_dn in sel_groups:
        attrs = groups.get(group_dn, None)
        if not attrs:
            continue
        res.append(group_dn)
        res.extend(get_subgroups(groups, attrs.get("member", [])))
    return res


def set_group_member_of(groups: dict, sel_groups: list, parents: list = None):
    if parents is None:
        parents = []

    for group_dn in sel_groups:
        attrs = groups.get(group_dn, None)
        if not attrs:
            continue
        attrs["memberOf"] = parents
        set_group_member_of(groups, attrs.get("member", []), parents + [group_dn])


def get_cn_or_ou_from_dn(dn: str) -> str:
    return dn.split(",")[0][3:]


def main():
    groups = read_ldap_json("groups.json")

    sel_roles = read_or_create_selected_groups("selected_roles.json", groups)
    sel_groups = read_or_create_selected_groups("selected_groups.json", groups)

    users = read_ldap_json("users.json")

    for user_dn, attrs in users.items():
        member_of_role = [groups[dn]["cn"] for dn in sel_roles if is_transitive_member_of(groups, attrs, dn)]
        if len(member_of_role) > 0:
            # "Benutzer_ID;Name;E-Mail;Rolle;Mitarbeiter;Benutzer_DN"
            # role = "Serviceberater" if "_SERV" in member_of[0] else "Buchhaltung"
            # print(
            #     ";".join(
            #         [
            #             attrs.get("sAMAccountName", ""),
            #             attrs.get("cn", ""),
            #             attrs.get("mail", ""),
            #             role,
            #             attrs.get("cn", ""),
            #             user_dn,
            #         ]
            #     )
            # )
            # print(attrs.get("cn", user_dn))
            for g in member_of_role:
                print(attrs.get("sAMAccountName", user_dn) + ";" + g)

            member_of_group = [
                get_cn_or_ou_from_dn(dn) for dn in sel_groups if is_transitive_member_of(groups, attrs, dn)
            ]
            if len(member_of_group) > 0:
                for g in member_of_group:
                    print(attrs.get("sAMAccountName", user_dn) + ";" + g)

    # for dn in sel_groups:
    #     group = groups.get(dn, {})
    #     print("\n" + dn, len(group.get("member", [])))

    #     for member_dn in group.get("member", []):
    #         print(" - " + member_dn)


def get_ou_subgroups(group_dn: str) -> list[str]:
    res = []
    group_dn_split = group_dn.split(",")
    for i, entry in enumerate(group_dn_split):
        if entry[:2] not in ("OU", "ou"):
            continue
        res.append(",".join(group_dn_split[i:]))
    return res


def is_transitive_member_of(groups: dict, dn_attrs: dict, group_dn: str) -> bool:
    if group_dn in get_ou_subgroups(dn_attrs["distinguishedName"]):
        return True

    for dn in dn_attrs.get("memberOf", []):
        if dn == group_dn:
            return True
        if dn in groups and is_transitive_member_of(groups, groups[dn], group_dn):
            return True
    return False


if __name__ == "__main__":
    main()