diff --git a/check_ad_2fa.py b/check_ad_2fa.py index 00208db..1f70e06 100644 --- a/check_ad_2fa.py +++ b/check_ad_2fa.py @@ -13,27 +13,26 @@ from ldap3 import Server, Connection, ALL, SUBTREE # Подтягиваем конфиг sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE +from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASES +from utils.ad_search import search_by_filter as ad_search_all_bases def find_user(login: str) -> dict | None: - """Найти пользователя в AD по логину (sAMAccountName или mail).""" - server = Server(AD_SERVER, get_info=ALL) - conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) - + """Найти пользователя в AD по логину (sAMAccountName или mail). + Ищет по всем OU из AD_BASES.""" filter_login = login.lower().replace("\\", "\\5c").replace("*", "\\2a").replace("(", "\\28").replace(")", "\\29") if "@" in filter_login: - ldap_filter = f"(|(mail={filter_login})(sAMAccountName={filter_login}))" + ldap_filter = "(|{}{})".format("(mail={})".format(filter_login), "(sAMAccountName={})".format(filter_login)) else: - ldap_filter = f"(|(sAMAccountName={filter_login})(mail={filter_login}))" + ldap_filter = "(|{}{})".format("(sAMAccountName={})".format(filter_login), "(mail={})".format(filter_login)) - conn.search(search_base=AD_BASE, search_filter=ldap_filter, attributes=["cn", "sAMAccountName", "memberOf"]) + entries = ad_search_all_bases(ldap_filter, ["cn", "sAMAccountName", "memberOf"]) - if not conn.entries: + if not entries: return None - entry = conn.entries[0] + entry = entries[0] return { "dn": entry.entry_dn, "cn": str(entry.cn), @@ -64,7 +63,7 @@ def main(): print(" Проверка пользователя в группе 2FA (Active Directory)") print("=" * 60) print(f"\n Сервер AD: {AD_SERVER}") - print(f" База поиска: {AD_BASE}") + print(f" Базы поиска: {config.AD_BASES}") print(f" Подключено как: {AD_USER}") print() diff --git a/config/config.py b/config/config.py index 38d1da5..cadcc4c 100644 --- a/config/config.py +++ b/config/config.py @@ -104,7 +104,10 @@ ALLOWED_USERS = [ # Список TrueConf ID разрешенны TC_SERVER = os.getenv("TC_SERVER", "") # Адрес сервера TrueConf AD_SERVER = "ldap://172.16.20.20" # Адрес контроллера домена Active Directory -AD_BASE = "OU=-Пользователи,DC=sibcem,DC=ru" # Базовый путь поиска пользователей в AD +AD_BASES = ( # Базовые пути поиска пользователей в AD (несколько OU) + "OU=-Пользователи,DC=sibcem,DC=ru", + "OU=Планшеты,OU=enabled,OU=БезКомпьютеров,DC=sibcem,DC=ru", +) SQL_SERVER = "SRVKEM-MOBILEIN.sibcem.ru" # Сервер MS SQL для работы Личного кабинета SQL_DB_NAME = "BossCopy" # База данных с информацией по кадрам и ЗП diff --git a/debug_ad_filter.py b/debug_ad_filter.py index d24b411..1a14f83 100644 --- a/debug_ad_filter.py +++ b/debug_ad_filter.py @@ -3,7 +3,8 @@ import sys sys.path.insert(0, "/opt/trueconf_bot") -from config.config import AD_USER, AD_PASSWORD, AD_SERVER, AD_BASE +from config.config import AD_USER, AD_PASSWORD, AD_SERVER, AD_BASES +from utils.ad_search import search_by_filter as ad_search_all_bases from ldap3.utils.conv import escape_filter_chars @@ -19,7 +20,14 @@ print(f"short_username: {short_username}") print(f"safe_user_id: {safe_user_id}") # lk/handlers.py filter -filter_lk = f"(&(objectClass=user)(|(mail={safe_user_id})(userPrincipalName={safe_user_id})(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')}))(sAMAccountName={escape_filter_chars(short_username)})))" +filter_lk = "(&(objectClass=user)(|(mail={})(userPrincipalName={})(userPrincipalName={}{}{}))({}))".format( + safe_user_id, + safe_user_id, + escape_filter_chars(short_username), + "@sibcem.ru", + "", + escape_filter_chars(short_username), +) print(f"\nFilter (lk format):") print(filter_lk) @@ -29,27 +37,23 @@ closes = filter_lk.count(')') print(f"\nOpen parens: {opens}, Close parens: {closes}") # Now test with the connection -from ldap3 import Server, Connection, ALL -server = Server(AD_SERVER, get_info=ALL) -conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) +print(f"\nSearching with filter across all AD_BASES...") +entries = ad_search_all_bases(filter_lk, ["cn", "sAMAccountName", "mail", "userPrincipalName"]) -print(f"\nSearching with filter...") -conn.search(AD_BASE, filter_lk, attributes=["cn", "sAMAccountName", "mail", "userPrincipalName"]) - -if conn.entries: - entry = conn.entries[0] +if entries: + entry = entries[0] print(f"FOUND: cn={entry.cn}, mail={entry.mail if 'mail' in entry else 'N/A'}, sam={entry.sAMAccountName if 'sAMAccountName' in entry else 'N/A'}") else: - print("NOT FOUND in AD_BASE") + print("NOT FOUND in any AD_BASE") # Try root -conn2 = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) root_base = "DC=sibcem,DC=ru" print(f"\nSearching in {root_base}...") -conn2.search(root_base, filter_lk, attributes=["cn", "sAMAccountName", "mail", "userPrincipalName"]) +from utils.ad_search import search_by_filter +entries2 = search_by_filter(filter_lk, ["cn", "sAMAccountName", "mail", "userPrincipalName"]) -if conn2.entries: - entry = conn2.entries[0] +if entries2: + entry = entries2[0] print(f"FOUND: cn={entry.cn}, mail={entry.mail if 'mail' in entry else 'N/A'}, sam={entry.sAMAccountName if 'sAMAccountName' in entry else 'N/A'}") else: print("NOT FOUND in root") diff --git a/instruct/handlers.py b/instruct/handlers.py index 190029e..5936029 100644 --- a/instruct/handlers.py +++ b/instruct/handlers.py @@ -21,15 +21,14 @@ logger = logging.getLogger(__name__) router = Router() -def _get_user_cn(user_id: str) -> str: +def get_user_cn(user_id: str) -> str | None: """ Получить CN пользователя из AD по его user_id (email или логин). user_id из TrueConf — это обычно email, например 'ds.krivochenko@tcs.sibcem.ru' Заменяет tcs.sibcem.ru на sibcem.ru для поиска в AD. + Ищет по всем OU из AD_BASES. """ - from ldap3 import Server, Connection, ALL - from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE - from ldap3.utils.conv import escape_filter_chars + from utils.ad_search import search_by_user_id try: # Заменяем tcs.sibcem.ru на sibcem.ru @@ -38,26 +37,10 @@ def _get_user_cn(user_id: str) -> str: logger.info(f"AD lookup: user_id={user_id}, search_id={search_id}, short={short_username}") - server = Server(AD_SERVER, get_info=ALL) - conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) + entries = search_by_user_id(search_id, ["cn", "sAMAccountName", "mail", "userPrincipalName"]) - # escape_filter_chars экранирует @ и . в email, что ломает фильтр. - # Для email и UPN передаём raw. Для sAMAccountName — экранируем. - safe_sam = escape_filter_chars(short_username) - - search_filter = ( - f"(&(objectClass=user)(|" - f"(mail={search_id})" - f"(userPrincipalName={search_id})" - f"(userPrincipalName={short_username}@sibcem.ru))" - f"(sAMAccountName={safe_sam}))" - ) - logger.info(f"AD filter: {search_filter}") - - conn.search(AD_BASE, search_filter, attributes=["cn", "sAMAccountName", "mail", "userPrincipalName"]) - - if conn.entries: - entry = conn.entries[0] + if entries: + entry = entries[0] cn = str(entry.cn) mail = str(entry.mail) if 'mail' in entry else "N/A" sam = str(entry.sAMAccountName) if 'sAMAccountName' in entry else "N/A" @@ -110,7 +93,7 @@ async def instruct_router_handler(msg: Message): elif cmd == "1": log_menu_stats(user_id, "Инструкции", "Trueconf") set_state(user_id, "INSTRUCT_TRUECONF_VIEW") - cn = _get_user_cn(user_id) + cn = get_user_cn(user_id) text = INSTRUCT_TRUECONF_WITH_AD(EMOJI_DIGITS, cn) await msg.answer(text, parse_mode="html") elif cmd == "2": diff --git a/lk/handlers.py b/lk/handlers.py index 8766d6b..289f954 100644 --- a/lk/handlers.py +++ b/lk/handlers.py @@ -129,19 +129,18 @@ async def handle_lk_error(msg: Message, user_id: str, action: str, error_details # ИНТЕГРАЦИЯ С ACTIVE DIRECTORY ЧЕРЕЗ extensionAttribute2 # ========================================================= def get_card_id_from_ad(user_id: str) -> str: + # Поиск CARD_ID по всем OU из AD_BASES + from utils.ad_search import search_by_user_id try: - server = Server(config.AD_SERVER, get_info=ALL) - conn = Connection(server, user=config.AD_USER, password=config.AD_PASSWORD, auto_bind=True) - safe_user_id = escape_filter_chars(user_id) - short_username = user_id.split('@')[0] - search_filter = f"(&(objectClass=user)(|(mail={safe_user_id})(userPrincipalName={safe_user_id})(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')})(sAMAccountName={escape_filter_chars(short_username)})))" - - conn.search(config.AD_BASE, search_filter, attributes=['extensionAttribute2']) - - if conn.entries and 'extensionAttribute2' in conn.entries[0] and conn.entries[0].extensionAttribute2.value: - return str(conn.entries[0].extensionAttribute2.value).strip() + entries = search_by_user_id(user_id, ['extensionAttribute2']) + if not entries: + raise Exception("Пользователь не найден в AD.") + entry = entries[0] + if 'extensionAttribute2' in entry and entry.extensionAttribute2.value: + return str(entry.extensionAttribute2.value).strip() raise Exception("Поле extensionAttribute2 не заполнено в AD.") - except Exception as e: raise Exception(f"AD Card Query Error: {e}") + except Exception as e: + raise Exception("AD Card Query Error: {}".format(e)) # ========================================================= # МОДНОЕ УНИВЕРСАЛЬНОЕ ЯДРО ПАГИНАЦИИ СПИСКОВ ДЛЯ ЧАТА diff --git a/lk/test_lk_payslip.py b/lk/test_lk_payslip.py index ec961f3..e2aac46 100644 --- a/lk/test_lk_payslip.py +++ b/lk/test_lk_payslip.py @@ -13,7 +13,10 @@ from ldap3.utils.conv import escape_filter_chars TEST_USER = "man.bogov@tcs.sibcem.ru" AD_SERVER = "ldap://172.16.20.20" -AD_BASE = "OU=-Пользователи,DC=sibcem,DC=ru" +AD_BASES = ( + "OU=-Пользователи,DC=sibcem,DC=ru", + "OU=Планшеты,OU=enabled,OU=БезКомпьютеров,DC=sibcem,DC=ru", +) AD_USER = "sdesk-mail" AD_PASSWORD = "fne?e!q.m8phcrGVAcqr" @@ -32,7 +35,6 @@ DB_PASSWORD = "Bav:fX#UwH8%atv4" def get_phone_from_ad(user_id: str) -> str: server = Server(AD_SERVER, get_info=ALL) - conn = Connection( server, user=AD_USER, @@ -41,22 +43,29 @@ def get_phone_from_ad(user_id: str) -> str: ) safe_user_id = escape_filter_chars(user_id) - short_username = user_id.split("@")[0] search_filter = ( - f"(&(objectClass=user)" - f"(|(mail={safe_user_id})" - f"(userPrincipalName={safe_user_id})" - f"(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')})" - f"(sAMAccountName={escape_filter_chars(short_username)})))" + "(&(objectClass=user)" + "(|(mail={})" + "(userPrincipalName={})" + "(userPrincipalName={}))" + "(sAMAccountName={}))".format( + safe_user_id, + safe_user_id, + escape_filter_chars(short_username) + "@sibcem.ru", + escape_filter_chars(short_username), + ) ) - conn.search( - AD_BASE, - search_filter, - attributes=['mobile', 'telephoneNumber'] - ) + for base in AD_BASES: + conn.search( + base, + search_filter, + attributes=['mobile', 'telephoneNumber'] + ) + if conn.entries: + break if not conn.entries: raise Exception("Пользователь не найден в AD") diff --git a/photo_bot/ad_helper.py b/photo_bot/ad_helper.py index 6d80fd9..c00072c 100644 --- a/photo_bot/ad_helper.py +++ b/photo_bot/ad_helper.py @@ -5,37 +5,39 @@ from ldap3 import Server, Connection, ALL, MODIFY_REPLACE # Подтягиваем конфиг sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE +from config.config import AD_SERVER, AD_USER, AD_PASSWORD +from utils.ad_search import search_by_login as ad_search_all_bases logger = logging.getLogger(__name__) def upload_photo_to_ad(target_user: str, photo_path: str) -> dict: """ Читает готовое фото с диска и записывает его в атрибут thumbnailPhoto в AD. + Ищет пользователя по всем OU из AD_BASES. """ + from utils.ad_search import search_by_filter try: # Читаем сырые байты картинки with open(photo_path, "rb") as f: photo_bytes = f.read() - # Формируем фильтр для поиска + # Ищем пользователя в AD по всем OU if '@' in target_user: - ldap_filter = f"(mail={target_user})" + ldap_filter = "(mail={})".format(target_user) else: - ldap_filter = f"(sAMAccountName={target_user})" + ldap_filter = "(sAMAccountName={})".format(target_user) - # Подключаемся к AD + entries = search_by_filter(ldap_filter, ["cn"]) + + if not entries: + return {"success": False, "error": "Пользователь {} не найден в AD.".format(target_user)} + + user_dn = entries[0].entry_dn + + # Подключаемся для модификации server = Server(AD_SERVER, get_info=ALL) conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) - # Находим пользователя в дереве (нам нужен его точный путь - DN) - conn.search(search_base=AD_BASE, search_filter=ldap_filter, attributes=["cn"]) - - if not conn.entries: - return {"success": False, "error": f"Пользователь {target_user} не найден в AD."} - - user_dn = conn.entries[0].entry_dn - # 🪄 МАГИЯ: Перезаписываем атрибут thumbnailPhoto нашими байтами conn.modify(user_dn, {'thumbnailPhoto': [(MODIFY_REPLACE, [photo_bytes])]}) diff --git a/photo_bot/check_ad_photo.py b/photo_bot/check_ad_photo.py index 26be3a8..a1f9263 100644 --- a/photo_bot/check_ad_photo.py +++ b/photo_bot/check_ad_photo.py @@ -7,7 +7,8 @@ from PIL import Image # Подтягиваем ваши настройки из конфига sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE +from config.config import AD_SERVER, AD_USER, AD_PASSWORD +from utils.ad_search import search_by_filter as ad_search_all_bases logging.basicConfig(level=logging.INFO, format="%(message)s") @@ -21,54 +22,46 @@ def check_user_photo(target_user: str): ldap_filter = f"(sAMAccountName={target_user})" try: - # Подключаемся к AD - server = Server(AD_SERVER, get_info=ALL) - conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) - - # Ищем пользователя и просим вернуть атрибуты: имя (cn) и фото (thumbnailPhoto) - conn.search( - search_base=AD_BASE, - search_filter=ldap_filter, - attributes=["cn", "thumbnailPhoto"] - ) - - if not conn.entries: + # Подключаемся к AD и ищем по всем OU из AD_BASES + entries = ad_search_all_bases(ldap_filter, ["cn", "thumbnailPhoto"]) + + if not entries: print("❌ Пользователь с такими данными не найден в AD.") return - - user = conn.entries[0] + + user = entries[0] user_name = user.cn.value if 'cn' in user else target_user print(f"👤 Найден сотрудник: {user_name}") - + # Проверяем, есть ли вообще фото if 'thumbnailPhoto' not in user or not user.thumbnailPhoto.value: print("⚠️ У этого пользователя НЕТ фотографии в Active Directory.") return - + # Достаем байты фотографии photo_bytes = user.thumbnailPhoto.value size_kb = len(photo_bytes) / 1024 - + # Читаем картинку прямо из памяти (без сохранения на диск) с помощью Pillow image = Image.open(io.BytesIO(photo_bytes)) width, height = image.size - + print("\n📊 --- ИНФОРМАЦИЯ О ФОТО В AD ---") print(f"📐 Разрешение : {width}x{height} пикселей") print(f"⚖️ Вес : {size_kb:.1f} KB") print(f"🗂 Формат : {image.format}") print("----------------------------------\n") - + # Бонус: сохраняем фото во временную папку, чтобы вы могли на него взглянуть safe_name = target_user.replace('@', '_').replace('.', '_') save_path = f"/tmp/{safe_name}_ad.jpg" - + with open(save_path, "wb") as f: f.write(photo_bytes) - + print(f"💾 Оригинал из AD сохранен сюда: {save_path}") print("Вы можете скачать его через WinSCP или открыть в TrueConf, чтобы оценить качество глазами.") - + except Exception as e: print(f"❌ Ошибка при подключении к AD: {e}") diff --git a/service_desk/handlers.py b/service_desk/handlers.py index 8f533da..997690e 100644 --- a/service_desk/handlers.py +++ b/service_desk/handlers.py @@ -70,23 +70,22 @@ sd_sessions = {} # ================================================ # БИЗНЕС-ЛОГИКА # ================================================ -def get_ad_user_sync(login: str): - try: - server = Server(AD_SERVER, get_info=ALL) - conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) - conn.search(search_base=AD_BASE, search_filter=f"(sAMAccountName={login})", attributes=["displayName", "mail", "l", "userAccountControl"]) - if conn.entries: - user = conn.entries[0] - uac = user.userAccountControl.value if 'userAccountControl' in user else 0 - return { - "name": user.displayName.value if 'displayName' in user else login, - "mail": user.mail.value if 'mail' in user else None, - "city": user.l.value if 'l' in user else "Кемерово", - "is_disabled": bool(uac & 2) - } - except Exception as e: - logger.error(f"Ошибка LDAP: {e}") - return None +def get_ad_user_sync(login: str): + try: + from utils.ad_search import search_by_login + entries = search_by_login(login, ["displayName", "mail", "l", "userAccountControl"]) + if entries: + user = entries[0] + uac = user.userAccountControl.value if 'userAccountControl' in user else 0 + return { + "name": user.displayName.value if 'displayName' in user else login, + "mail": user.mail.value if 'mail' in user else None, + "city": user.l.value if 'l' in user else "Кемерово", + "is_disabled": bool(uac & 2) + } + except Exception as e: + logger.error(f"Ошибка LDAP: {e}") + return None async def generate_smart_subject(text: str) -> str: import re, traceback, importlib.util, httpx diff --git a/transcription_bot/handlers.py b/transcription_bot/handlers.py index 73cd146..d364f90 100644 --- a/transcription_bot/handlers.py +++ b/transcription_bot/handlers.py @@ -45,11 +45,10 @@ ALLOWED_EXTENSIONS = [ # --- Функция получения Email из Active Directory --- def get_user_email_sync(login: str) -> str: try: - server = Server(AD_SERVER, get_info=ALL) - conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) - conn.search(search_base=AD_BASE, search_filter=f"(sAMAccountName={login})", attributes=["mail"]) - if conn.entries and 'mail' in conn.entries[0] and conn.entries[0].mail.value: - return str(conn.entries[0].mail.value) + from utils.ad_search import search_by_login + entries = search_by_login(login, ["mail"]) + if entries and 'mail' in entries[0] and entries[0].mail.value: + return str(entries[0].mail.value) except Exception as e: logger.error(f"Ошибка поиска email в AD: {e}") return DEFAULT_REQUESTER diff --git a/utils/ad_checker.py b/utils/ad_checker.py index 484c663..5abd900 100644 --- a/utils/ad_checker.py +++ b/utils/ad_checker.py @@ -3,34 +3,20 @@ from ldap3 import Server, Connection, ALL - - -def _get_ad_connection(): - """Создать соединение с AD.""" - from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE # noqa: E501 - server = Server(AD_SERVER, get_info=ALL) - return Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True), AD_BASE +from utils.ad_search import search_by_filter as ad_search_all_bases def get_user_groups(cn: str) -> list[str] | None: """ Найти пользователя по CN и вернуть список групп. + Ищет по всем OU из AD_BASES. Возвращает None если пользователь не найден. """ - conn, ad_base = _get_ad_connection() - try: - conn.search( - search_base=ad_base, - search_filter=f"(cn={cn})", - attributes=["cn", "sAMAccountName", "memberOf"], - ) - if not conn.entries: - return None - entry = conn.entries[0] - return [str(g) for g in entry.memberOf] - except Exception as e: - print(f"❌ Ошибка AD при поиске {cn}: {e}") + entries = ad_search_all_bases("(cn={})".format(cn), ["cn", "sAMAccountName", "memberOf"]) + if not entries: return None + entry = entries[0] + return [str(g) for g in entry.memberOf] def check_group_membership(cn: str, group_pattern: str = "2FA") -> dict: diff --git a/utils/ad_search.py b/utils/ad_search.py new file mode 100644 index 0000000..4b1e973 --- /dev/null +++ b/utils/ad_search.py @@ -0,0 +1,73 @@ +# /opt/trueconf_bot/utils/ad_search.py +# Универсальный поиск пользователей по нескольким OU в AD. +# Каждая функция принимает user_id/email/логин и ищет по всем OU из AD_BASES. +# Возвращает первую найденную запись (ldap3 Entry) или None. + +from ldap3 import Server, Connection, ALL +from ldap3.utils.conv import escape_filter_chars +import config.config as config + + +def _get_conn(): + """Создать соединение с AD.""" + server = Server(config.AD_SERVER, get_info=ALL) + return Connection(server, user=config.AD_USER, password=config.AD_PASSWORD, auto_bind=True) + + +def search_by_login(login: str, attributes: list[str] | None = None) -> list: + """ + Найти пользователя по sAMAccountName во всех OU из AD_BASES. + Возвращает список Entry (обычно 0 или 1). + """ + attrs = attributes or ["cn", "sAMAccountName"] + conn = _get_conn() + for base in config.AD_BASES: + conn.search( + search_base=base, + search_filter="({}={})".format("sAMAccountName", login), + attributes=attrs, + ) + if conn.entries: + return list(conn.entries) + return [] + + +def search_by_user_id(user_id: str, attributes: list[str] | None = None) -> list: + """ + Найти пользователя по email/UPN/sAMAccountName во всех OU из AD_BASES. + Возвращает список Entry (обычно 0 или 1). + """ + attrs = attributes or ["cn", "sAMAccountName", "mail", "userPrincipalName", "extensionAttribute2"] + safe_id = escape_filter_chars(user_id) + short_username = user_id.split("@")[0] if "@" in user_id else user_id + safe_sam = escape_filter_chars(short_username) + + # Строим LDAP filter: (&(objectClass=user)(|(mail=...)(userPrincipalName=...)(sAMAccountName=...))) + or_parts = [ + "(mail={})".format(safe_id), + "(userPrincipalName={})".format(safe_id), + "(userPrincipalName={})".format(short_username + "@sibcem.ru"), + "(sAMAccountName={})".format(safe_sam), + ] + or_filter = "({})".format("".join(or_parts)) + search_filter = "(&{}{})".format("(objectClass=user)", or_filter) + + conn = _get_conn() + for base in config.AD_BASES: + conn.search(search_base=base, search_filter=search_filter, attributes=attrs) + if conn.entries: + return list(conn.entries) + return [] + + +def search_by_filter(search_filter: str, attributes: list[str]) -> list: + """ + Найти по произвольному LDAP-фильтру во всех OU из AD_BASES. + Возвращает список Entry. + """ + conn = _get_conn() + for base in config.AD_BASES: + conn.search(search_base=base, search_filter=search_filter, attributes=attributes) + if conn.entries: + return list(conn.entries) + return [] diff --git a/utils/otp_service.py b/utils/otp_service.py index d3bdf65..d0ce857 100644 --- a/utils/otp_service.py +++ b/utils/otp_service.py @@ -63,18 +63,16 @@ async def send_otp_via_api(phone: str, code: str) -> tuple: return False, str(e) def get_card_id_from_ad(user_id: str) -> str: - """Синхронно вытягивает CARD_ID сотрудника из Active Directory (extensionAttribute2)""" + """Синхронно вытягивает CARD_ID сотрудника из Active Directory (extensionAttribute2). + Ищет по всем OU из AD_BASES.""" + from utils.ad_search import search_by_user_id try: - server = Server(config.AD_SERVER, get_info=ALL) - conn = Connection(server, user=config.AD_USER, password=config.AD_PASSWORD, auto_bind=True) - safe_user_id = escape_filter_chars(user_id) - short_username = user_id.split('@')[0] - - search_filter = f"(&(objectClass=user)(|(mail={safe_user_id})(userPrincipalName={safe_user_id})(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')})(sAMAccountName={escape_filter_chars(short_username)})))" - conn.search(config.AD_BASE, search_filter, attributes=['extensionAttribute2']) - - if conn.entries and 'extensionAttribute2' in conn.entries[0] and conn.entries[0].extensionAttribute2.value: - return str(conn.entries[0].extensionAttribute2.value).strip() + entries = search_by_user_id(user_id, ["extensionAttribute2"]) + if not entries: + raise Exception("Пользователь не найден в AD.") + entry = entries[0] + if 'extensionAttribute2' in entry and entry.extensionAttribute2.value: + return str(entry.extensionAttribute2.value).strip() raise Exception("Поле extensionAttribute2 не заполнено в AD.") except Exception as e: logger.error(f"Ошибка получения CARD_ID из AD для {user_id}: {e}")