res_levis_API/routes/majornet.py

from fastapi import APIRouter, HTTPException
from fastapi.param_functions import Depends
from fastapi.security import OAuth2PasswordRequestForm
from fastapi_login.exceptions import InvalidCredentialsException



import xml.dom.minidom
import xml.etree.ElementTree as ET
from binascii import unhexlify
from ldap3.protocol.formatters.formatters import format_sid
import argparse
import json
import ldap3
import logging
import os
import ssl
import sys
import hashlib
import binascii
from datetime import datetime, timedelta
import subprocess
from typing import List

#from db import get_session
from core.actions import get_user_by_name
from core.security import manager
from core.ldap import LDAPConsole
from security import get_current_user

from models.httpresponse import httpResponse400, httpResponse200, httpResponse500
from models.mnuser import mnuser, post_mnuser, mnuser_groups

router = APIRouter(
    prefix="/majornet"
)

MNUSERS_XML_PATH = os.getenv(
    "MNUSERS_XML_PATH",
    os.getenv("MNUSERS_XML", "/conf/etc/useradmin/mnusers.xml"),
)

def cast_to_dict(cid):
    out = {}
    for key, value in cid.items():
        if type(value) == bytes:
            out[key] = str(value)
        elif type(value) == list:
            if len(value) == 1:
                value = value[0]
                if type(value) == bytes:
                    out[key] = str(value)
                elif type(value) == datetime:
                    out[key] = value.strftime('%Y-%m-%d %T')
                elif type(value) == timedelta:
                    # Output format to change
                    out[key] = value.seconds
                else:
                    out[key] = value
            else:
                newlist = []
                for element in value:
                    if type(element) == bytes:
                        newlist.append(str(element))
                    elif type(element) == datetime:
                        newlist.append(element.strftime('%Y-%m-%d %T'))
                    elif type(element) == timedelta:
                        # Output format to change
                        newlist.append(element.seconds)
                out[key] = newlist
        elif type(value) == datetime:
            out[key] = value.strftime('%Y-%m-%d %T')
        elif type(value) == timedelta:
            # Output format to change
            out[key] = value.seconds
        else:
            out[key] = value
    return out


def dict_get_paths(d):
    paths = []
    for key in d.keys():
        if type(d[key]) == dict:
            paths = [[key]+p for p in dict_get_paths(d[key])]
        else:
            paths.append([key])
    return paths


def dict_path_access(d, path):
    for key in path:
        if key in d.keys():
            d = d[key]
        else:
            return None
    return d


def _element_text(element, tag_name):
    child = element.find(tag_name)
    if child is None or child.text is None:
        return ""
    return child.text.strip()


def _split_usergroups(value):
    if not value:
        return []
    return [group.strip() for group in value.split(",") if group.strip()]


def _iter_mnuser_nodes(root):
    user_nodes = root.findall(".//useradmin")
    if user_nodes:
        return user_nodes
    return [node for node in root.iter() if node.find("uid") is not None]


def read_mnusers_groups(file_path=MNUSERS_XML_PATH):
    tree = ET.parse(file_path)
    root = tree.getroot()
    users = []

    for node in _iter_mnuser_nodes(root):
        uid = _element_text(node, "uid")
        if not uid:
            continue
        users.append({
            "uid": uid,
            "usergroups": _split_usergroups(_element_text(node, "usergroups")),
        })

    return users










data = {}    
import os
USE_LDAP = os.getenv("FASTAPI_LDAP_ENABLED", "false").lower() == "true"

try:
    from ..core.ldap import LDAPConsole
except Exception:
    LDAPConsole = None

lc = LDAPConsole(debug=True) if (USE_LDAP and LDAPConsole) else None



@router.get(
    "/getUsersGroups",
    response_model=List[mnuser_groups],
    tags=["MajorNet"],
    dependencies=[Depends(get_current_user)],
)
async def get_users_groups():
    try:
        return read_mnusers_groups()
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail="mnusers.xml not found")
    except ET.ParseError:
        raise HTTPException(status_code=500, detail="Invalid mnusers.xml")
    except OSError:
        raise HTTPException(status_code=500, detail="Unable to read mnusers.xml")


@router.get("/users/",tags=["MajorNet"], responses={200: {"model": httpResponse200}, 400: {"model": httpResponse400}, 500: {"model": httpResponse500}})
async def get_majornet_users(current_user= Depends(manager)):
    
    response = lc.queryallusers("ou=users", attributes=['displayName','mail','uid'])
    data = {}
    for cn in response:
            path = cn.split(',')[::-1]
            tmp = data
            for key in path[:-1]:
                if key in tmp.keys():
                    tmp = tmp[key]
                else:
                    tmp[key] = {}
                    tmp = tmp[key]
            tmp[path[-1]] = cast_to_dict(response[cn])

    json_data = json.dumps(data, indent=4)
    return data

#,current_user= Depends(manager)
#,current_user:  Depends(manager)
# response_model=post_mnuser,
@router.post("/users/add_user/", response_model=post_mnuser, tags=["MajorNet"])
async def majornet_add_user(mnuser:str, mnpasswd:str, mndisplayname:str, mnmail="", mnprofile= "default",current_user= Depends(manager) ):
    #print (current_user.username)
    #print (mnuser+" "+mnpasswd+" "+mndisplayname+" "+mnprofile)
    proc = subprocess.Popen(['perl', '/var/opt/FastAPI/addusr.pl', mnuser, mnpasswd , mndisplayname, mnmail ], stdout=subprocess.PIPE)
    stdout_value = proc.communicate()[0]
    
    stdout_value=stdout_value.decode('UTF-8')
    response = {"return_code": stdout_value.split(';')[0], "return_str" : stdout_value.split(';')[1] }
    return response