Python-Projekte/ldap/convert_ldap_to_db.py

import json
from dataclasses import astuple, dataclass
from pathlib import Path

base_dir = Path(__file__).parent / "Reisacher"


@dataclass
class Benutzer:
    Benutzer_ID: str
    Name: str
    Email: str
    Rolle: str
    Mitarbeiter: str
    Benutzer_DN: str


@dataclass
class Benutzer_Gruppe:
    Benutzer_ID: str
    Gruppe_ID: str


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]
        sel_groups.sort()
        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")
    users_db = []
    users_roles = {}
    users_groups = {}

    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:
            continue

        role = "Serviceberater" if "serv" in member_of_role[0].lower() else "Buchhaltung"
        users_db.append(
            Benutzer(
                Benutzer_ID=attrs.get("sAMAccountName", ""),
                Name=attrs.get("cn", ""),
                Email=attrs.get("mail", ""),
                Rolle=role,
                Mitarbeiter=attrs.get("cn", ""),
                Benutzer_DN=user_dn,
            )
        )
        user_id = attrs.get("sAMAccountName", user_dn)
        users_roles[user_id] = member_of_role
        users_groups[user_id] = [
            get_cn_or_ou_from_dn(dn) for dn in sel_groups if is_transitive_member_of(groups, attrs, dn)
        ]

    with open(base_dir / "insert_users_and_groups.sql", "w", encoding="latin-1") as fwh:
        for role in sel_roles:
            role_info = (groups[role].get("cn", ""), groups[role].get("cn", ""), role, "1")
            fwh.write("INSERT INTO [auth].[Rollen] VALUES ('" + "', '".join(role_info) + "');\n")

        fwh.write("\n")

        for group in sel_groups:
            group_info = (groups[group].get("cn", ""), groups[group].get("cn", ""), group)
            fwh.write(
                "INSERT INTO [auth].[Gruppen] (Gruppe_ID, Gruppe_Name, Gruppe_DN) VALUES ('"
                + "', '".join(group_info)
                + "');\n"
            )

        fwh.write("\n\n")

        for user in users_db:
            fwh.write("INSERT INTO [auth].[Benutzer] VALUES ('" + "', '".join(astuple(user)) + "');\n")

        fwh.write("\n")

        for user_id, roles in users_roles.items():
            for role in roles:
                fwh.write("INSERT INTO [auth].[Benutzer_Gruppe] VALUES ('" + user_id + "', '" + role + "');\n")

        fwh.write("\n")

        for user_id, groups in users_groups.items():
            for group in groups:
                fwh.write("INSERT INTO [auth].[Benutzer_Gruppe] VALUES ('" + user_id + "', '" + group + "');\n")

    # 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()