dkovtun
/
Lennergo_migrate_script
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

добавлен Скрипт для отображения Организационный структуры пользователя а также внесены изменения скрипт создания пользователя

main
Dmitry Kovtun 2024-06-07 10:56:28 +03:00
parent 3558b32803
commit 5813e93ddc
4 changed files with 128 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
create_ou_branch.py
View File

@ -136,8 +136,11 @@ if __name__ == "__main__":
manager = Manager(smb, ad)
date_file = manager.open_csv_file(PATH_CSV)
if date_file is not None:
ou_add_list = [ i[5] for i in date_file]
ou_comp = [i[10] for i in date_file]
try:
ou_add_list = [ i[5] for i in date_file]
ou_comp = [i[10] for i in date_file]
except Exception as ex:
logging.error(ex)
logging.info("run create ou")
# manager.creat_branch_ou(ou_add_list)# Создаст структуру OU
manager.creat_branch_ou(ou_add_list)

9
create_users.py
View File

@ -99,10 +99,8 @@ class Samba_provaider():
cmd.append("--internet-address={}".format(data.get("wWWHomePage")))
if data.get("initials"):
cmd.append("--initials={}".format(data.get("initials")))
# if data.get("homeDirectory"):
# cmd.append("--home-directory={}".format(data.get("homeDirectory")))
cmd.append("--must-change-at-next-login")
cmd.append("--use-username-as-cn")
# cmd.append("--use-username-as-cn")
cmd.append( "--userou={}".format(ou))
out = subprocess.call(cmd,restore_signals=True, shell=False)
if out == 0:
@ -160,6 +158,9 @@ class Samba_provaider():
if data.get("co"):
out_str.append("add: co")
out_str.append("co: {}".format(data.get("co")))
# if data.get("cn"):
# out_str.append("add: cn")
# out_str.append("cn: {}".format(data.get("cn")))
if data.get("mobile"):
out_str.append("add: mobile")
out_str.append("mobile: {}".format(data.get("mobile")))
@ -180,7 +181,7 @@ class Samba_provaider():
with open("/tmp/{}.ldif".format(user), "w") as file:
file.write("\n".join(out_str))
out = subprocess.call(cmd,restore_signals=True, shell=False)
print(out)
# print(out)
if out == 0:
return True
return False

52
find_ms_group.py Normal file
View File

@ -0,0 +1,52 @@
#!/usr/bin/python3
from ldap3 import Server, Connection, ALL, NTLM, SUBTREE, SAFE_SYNC, BASE
# from samba.samdb import SamDB
# from samba.auth import system_session
# from samba.ndr import ndr_pack, ndr_unpack
# from samba.dcerpc import security
# import samba.param
import logging
from pprint import pprint
import json
import csv
import subprocess
import time
class AD_provaider():
def __init__(self, url:str, serch_tree:str, user, password) -> None:
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
self.__server = Server(url)
self.__connect = Connection(self.__server, user, password)
self.__ad_serch_tree = serch_tree
if self.__connect.bind():
logging.info("status connect AD.........ok")
else:
logging.warning("status connect AD.........error")
def search_ms_ad(self, search_filter ,filter:list = ["*"])->dict:
logging.info("search >>>>>>>>>>>>>> AD")
self.__connect.search(self.__ad_serch_tree, search_filter, SUBTREE, attributes=filter)
response = self.__connect.response_to_json()
response = json.loads(response)
response = json.dumps(response, ensure_ascii="utf-8")
out = json.loads(response)
pprint(out, indent=4)
# return json.loads(response)
MS_AD_ADRESS = 'ldap://cp-vm-dc01.energo.ru'
SEARCH_FREE_MS = "dc=energo,dc=ru"
MS_USER = 'energo\\administrator'
PASSWORD = "P@sww0rd"
PATH_SCV = "List_groups.csv"
SEARCH_FREE_SAMBA = "dc=lenenergo,dc=ru"
ad = AD_provaider(MS_AD_ADRESS, SEARCH_FREE_MS, MS_USER, PASSWORD)
# ad.search_ms_ad("(&(objectCategory=group)(name=test)(distinguishedName=CN=test,ou,base))") # Поиск

66
get_users_data.py Normal file
View File

@ -0,0 +1,66 @@
from ldap3 import Server, Connection, ALL, NTLM, SUBTREE, SAFE_SYNC, BASE
import logging
import json
from pprint import pprint
class AD_provaider():
def __init__(self, url:str, serch_tree:str, user, password) -> None:
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
self.__server = Server(url)
self.__connect = Connection(self.__server, user, password, authentication=NTLM)
self.__ad_serch_tree = serch_tree
if self.__connect.bind():
logging.info("status connect AD.........ok")
else:
logging.warning("status connect AD.........error")
def search_ms_ad(self,search_filter ,filter:list = ["*"], dn = None)->dict:
logging.info("search >>>>>>>>>>>>>> AD")
if dn is not None:
self.__connect.search(dn, search_filter, SUBTREE, attributes=filter)
else:
self.__connect.search(self.__ad_serch_tree, search_filter, SUBTREE, attributes=filter)
response = self.__connect.response_to_json()
response = json.loads(response)
response = json.dumps(response, ensure_ascii="utf-8")
return json.loads(response)
if __name__ == "__main__":
MS_AD_ADRESS = 'ldap://cp-vm-dc01.energo.ru'
SEARCH_FREE_MS = "dc=energo,dc=ru"
MS_USER = 'energo\\administrator'
PASSWORD = "P@sww0rd"
ad = AD_provaider(MS_AD_ADRESS, SEARCH_FREE_MS, MS_USER, PASSWORD)
#################################################################
users_list = [
"bin",
"test",
"test_01"
] # Здесь находятся список пользователей по которому мы ищем
##################################################################
users = {}
users["Users"] = []
for user in users_list:
data = ad.search_ms_ad(f"(sAMAccountName={user})")
user_js = data.get("entries")[0].get("attributes")
out = {}
out["sAMAccountName"] = user_js.get("sAMAccountName")
out["email"] = user_js.get("mail")
# out["groups"] = user_js.get("memberOf")
ou = [i for i in user_js.get("distinguishedName").split(",") if i.split("=")[0] != "CN" ]
out["ou"] = ",".join(str(ou).encode("utf-8"))
p = json.dumps(out, ensure_ascii=False)
dumps = json.loads(p)
users["Users"].append(out)
with open("data_users.json", "w", encoding="utf-8") as f:
json.dump(users, f, indent=4, ensure_ascii=False)