From 7bd7700314a8bf387dc99dd07c7d0667d34fa2d3 Mon Sep 17 00:00:00 2001 From: ohmo Date: Sat, 4 Jul 2026 13:33:57 +0700 Subject: [PATCH] refactor: use Passwork for AD credentials (like prod config) --- ad/test_ad_lookup.py | 219 ++++++++++++++++++++++++------------------- 1 file changed, 122 insertions(+), 97 deletions(-) diff --git a/ad/test_ad_lookup.py b/ad/test_ad_lookup.py index f078d75..b08ef37 100644 --- a/ad/test_ad_lookup.py +++ b/ad/test_ad_lookup.py @@ -1,59 +1,88 @@ #!/usr/bin/env python3 """ -Тестовый скрипт для проверки обращения к Active Directory. -Аналог find_user_in_ad() из sd_dispatcher.py — standalone. +Тестовый скрипт — поиск пользователя в Active Directory. +Аналог find_user_in_ad() из sd_dispatcher.py. -Установка зависимостей (на сервере 192.168.1.106): - python3 -m pip install ldap3 +Использует Passwork для загрузки учётных данных AD (как в основном коде). +Если Passwork недоступен — просит ввести учётные данные вручную. -Использование: - # 1) Через аргументы: - python3 test_ad_lookup.py --user "sibcem\\svc_dispatcher" --pass "password" --login ivanov +Единственная зависимость: ldap3 + pip install ldap3 - # 2) Через env-переменные: - export AD_USER="sibcem\\svc_dispatcher" - export AD_PASS="password" +Запуск: python3 test_ad_lookup.py --login ivanov - - # 3) По email: python3 test_ad_lookup.py --email ivanov@sibcem.ru - - # 4) По телефону: python3 test_ad_lookup.py --phone 79131234567 - - # 5) Тест: найти первого попавшегося пользователя и проверить все методы: python3 test_ad_lookup.py --test-all """ import argparse +import base64 +import json import os import re import sys -# --------------------------------------------------------------------------- -# 1. Загрузка учётных данных -# --------------------------------------------------------------------------- -def get_credentials(): - """Возвращает (user, password, base_dn).""" - user = os.getenv("AD_USER", "").strip() - password = os.getenv("AD_PASS", "").strip() - base = os.getenv("AD_BASE", "").strip() or "OU=-Пользователи,DC=sibcem,DC=ru" +# ========================================================================= +# 1. Загрузка учётных данных AD +# ========================================================================= - return user, password, base +def load_ad_credentials(): + """ + Загружает AD_USER и AD_PASSWORD из Passwork (как в config.py). + fallback: ручной ввод. + """ + # --- Путь 1: через модуль Passwork --- + passwork_dir = "/opt/passwork" + card_name = os.getenv("PW_ID_AD", "AD").strip() + + if passwork_dir not in sys.path: + sys.path.insert(0, passwork_dir) + + try: + from passwork import get_passwork_secrets + print(f"[Passwork] Загрузка карточки '{card_name}'...") + pool = get_passwork_secrets(required_cards=[card_name]) + card = pool.get(card_name) + if card: + ad_user = card.get("login", "").strip() + ad_pass = card.get("password", "").strip() + if ad_user and ad_pass: + print(f"[Passwork] ✅ AD_USER: {ad_user}") + return ad_user, ad_pass + except Exception: + pass + + # --- Путь 2: env-переменные --- + ad_user = os.getenv("AD_USER", "").strip() + ad_pass = os.getenv("AD_PASS", "").strip() + if ad_user and ad_pass: + print("[Env] ✅ Учётные данные загружены из окружения") + return ad_user, ad_pass + + # --- Путь 3: ручной ввод --- + print("\n❌ Passwork недоступен.") + print("Введите учётные данные AD вручную:\n") + ad_user = input(" AD user (DOMAIN\\account): ").strip() + ad_pass = input(" AD password: ").strip() + if not ad_user or not ad_pass: + print("❌ Отмена.") + sys.exit(1) + print(f"[Manual] ✅ AD_USER: {ad_user}") + return ad_user, ad_pass -# --------------------------------------------------------------------------- +# ========================================================================= # 2. Подключение к LDAP -# --------------------------------------------------------------------------- +# ========================================================================= -def connect_to_ad(ad_user, ad_pass, ad_base): - """Возвращает подключённый ldap3.Connection.""" +def connect_to_ad(ad_user, ad_pass): from ldap3 import Server, Connection server = Server( "ldap://172.16.20.20", - get_info=None, # не тянем schema — экономим время + get_info=None, connect_timeout=5, ) conn = Connection( @@ -63,17 +92,17 @@ def connect_to_ad(ad_user, ad_pass, ad_base): auto_bind=True, receive_timeout=5, ) - return conn, ad_base + return conn -# --------------------------------------------------------------------------- +# ========================================================================= # 3. Поиск пользователя -# --------------------------------------------------------------------------- +# ========================================================================= -def find_user(conn, ad_base, login=None, email=None, phone=None): +def find_user(conn, login=None, email=None, phone=None): """ - Ищет пользователя в AD. Приоритет: телефон > логин > email. - Возвращает dict с атрибутами или None. + Приоритет: телефон > логин > email. + Возвращает dict атрибутов или None. """ if phone: filt = ( @@ -93,20 +122,21 @@ def find_user(conn, ad_base, login=None, email=None, phone=None): print("❌ Укажите --login, --email или --phone") sys.exit(1) - print(f"\n[LDAP] Поиск: {filt}") - print(f"[LDAP] Base DN: {ad_base}") + print(f"\n[LDAP] Фильтр: {filt}") + print(f"[LDAP] Base: OU=-Пользователи,DC=sibcem,DC=ru") - # Запрос 1 — основной conn.search( - search_base=ad_base, + search_base="OU=-Пользователи,DC=sibcem,DC=ru", search_filter=filt, - attributes=["l", "company", "manager", "department", "ou", - "displayName", "mail", "sAMAccountName", "title", - "telephoneNumber", "mobile", "givenName", "sn"], + attributes=[ + "sAMAccountName", "displayName", "givenName", "sn", + "mail", "title", "department", "l", "company", + "manager", "telephoneNumber", "mobile", + ], ) if not conn.entries: - print(f"\n❌ Пользователь не найден по {by}") + print(f"\n❌ Не найден по {by}") return None user = conn.entries[0] @@ -127,16 +157,16 @@ def find_user(conn, ad_base, login=None, email=None, phone=None): "mobile": user.mobile.value if "mobile" in user else None, } - # Город из OU-пути если атрибут l пустой + # Город из OU если атрибут l пустой if not result["city"]: match = re.search(r"OU=([^,]+),OU=-Пользователи", result["dn"], re.IGNORECASE) if match: result["city"] = match.group(1).strip() if result["city"].lower() == "красноярск" and "OU=ООО Комбинат Волна" in result["dn"]: result["company"] = "ООО Комбинат Волна" - print(f" Город извлечён из OU: {result['city']}") + print(f" Город из OU: {result['city']}") - # Запрос 2 — email руководителя + # Руководитель — второй запрос manager_dn = user.manager.value if "manager" in user else None if manager_dn: conn.search( @@ -147,7 +177,9 @@ def find_user(conn, ad_base, login=None, email=None, phone=None): ) if conn.entries and "mail" in conn.entries[0]: result["manager_email"] = conn.entries[0].mail.value - result["manager_name"] = conn.entries[0].displayName.value if "displayName" in conn.entries[0] else None + result["manager_name"] = ( + conn.entries[0].displayName.value if "displayName" in conn.entries[0] else None + ) else: result["manager_email"] = None else: @@ -156,12 +188,11 @@ def find_user(conn, ad_base, login=None, email=None, phone=None): return result -# --------------------------------------------------------------------------- -# 4. Форматирование вывода -# --------------------------------------------------------------------------- +# ========================================================================= +# 4. Вывод +# ========================================================================= def print_report(data): - """Красивый вывод результатов.""" print("\n" + "=" * 60) print(" РЕЗУЛЬТАТ ПОИСКА В ACTIVE DIRECTORY") print("=" * 60) @@ -183,38 +214,33 @@ def print_report(data): print(f" DN: {data['dn']}") # OU-дерево - dn = data["dn"] - parts = [p.strip() for p in dn.split(",")] - ou_parts = [p for p in parts if p.startswith("OU=")] + ou_parts = [p.strip() for p in data["dn"].split(",") if p.strip().startswith("OU=")] if ou_parts: print(f"\n OU-структура:") for i, ou in enumerate(reversed(ou_parts)): - indent = " " * i - print(f" {indent}└── {ou}") + print(f" {' ' * i}└── {ou}") print("=" * 60) -# --------------------------------------------------------------------------- -# 5. Тестирование всех методов поиска -# --------------------------------------------------------------------------- +# ========================================================================= +# 5. Тест всех методов +# ========================================================================= -def test_all_methods(conn, ad_base): - """Находит первого пользователя и проверяет все методы поиска.""" +def test_all_methods(conn): print("\n" + "─" * 60) - print(" РЕЖИМ ТЕСТИРОВАНИЯ: поиск по всем методам") + print(" ТЕСТИРОВАНИЕ: все методы поиска") print("─" * 60) conn.search( - search_base=ad_base, + search_base="OU=-Пользователи,DC=sibcem,DC=ru", search_filter="(objectClass=user)", - attributes=["sAMAccountName", "mail", "telephoneNumber", "displayName", - "l", "company", "department"], + attributes=["sAMAccountName", "mail", "telephoneNumber"], size_limit=1, ) if not conn.entries: - print("❌ В AD нет пользователей.") + print("❌ Нет пользователей в AD.") return sample = conn.entries[0] @@ -222,61 +248,60 @@ def test_all_methods(conn, ad_base): mail = sample.mail.value if "mail" in sample else None phone = sample.telephoneNumber.value if "telephoneNumber" in sample else None - print(f"\n📌 Тестовый пользователь: {sam} ({mail})") + print(f"\n📌 Тестовый: {sam} ({mail})") for label, kwargs in [("логину", {"login": sam}), ("email", {"email": mail}) if mail else None, ("телефону", {"phone": phone}) if phone else None]: if kwargs is None: continue - print(f"\n[Тест] Поиск по {label}...") - r = find_user(conn, ad_base, **kwargs) + print(f"\n[Тест] По {label}...") + r = find_user(conn, **kwargs) if r: print_report(r) print("\n" + "─" * 60) - print(" ✅ Тестирование завершено") + print(" ✅ Готово") print("─" * 60) -# --------------------------------------------------------------------------- +# ========================================================================= # MAIN -# --------------------------------------------------------------------------- +# ========================================================================= if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Тестовый поиск пользователя в AD") - parser.add_argument("--user", help="LDAP user (sAMAccountName или DN)") - parser.add_argument("--pass", dest="password", help="LDAP password") - parser.add_argument("--base", help="Base DN (по умолчанию OU=-Пользователи,DC=sibcem,DC=ru)") + parser = argparse.ArgumentParser( + description="Поиск пользователя в Active Directory", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Примеры: + python3 test_ad_lookup.py --login ivanov + python3 test_ad_lookup.py --email ivanov@sibcem.ru + python3 test_ad_lookup.py --phone 79131234567 + python3 test_ad_lookup.py --test-all + +Учётные данные AD загружаются из Passwork (как в основном коде). +Если Passwork недоступен — запрашиваются вручную. + """, + ) group = parser.add_mutually_exclusive_group(required=True) - group.add_argument("--login", help="sAMAccountName пользователя для поиска") - group.add_argument("--email", help="Email пользователя для поиска") - group.add_argument("--phone", help="Номер телефона для поиска") + group.add_argument("--login", help="sAMAccountName пользователя") + group.add_argument("--email", help="Email пользователя") + group.add_argument("--phone", help="Номер телефона") parser.add_argument("--test-all", action="store_true", - help="Найти первого пользователя и протестировать все методы поиска") + help="Найти первого пользователя и проверить все методы") args = parser.parse_args() - # Учётные данные: аргументы > env > prompt - ad_user = args.user or os.getenv("AD_USER", "") - ad_pass = args.password or os.getenv("AD_PASS", "") - ad_base = args.base or os.getenv("AD_BASE", "OU=-Пользователи,DC=sibcem,DC=ru") + # Загрузка учётных данных (Passwork → env → ручной ввод) + ad_user, ad_pass = load_ad_credentials() - if not ad_user or not ad_pass: - # Пробуем получить из .env файла проекта - env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "sd_dispatcher", "config", ".env") - if os.path.exists(env_path): - print(f"[.env] Загружено: {env_path}") - else: - print("❌ Укажите --user и --pass, или задайте AD_USER / AD_PASS") - sys.exit(1) - - conn, ad_base = connect_to_ad(ad_user, ad_pass, ad_base) + conn = connect_to_ad(ad_user, ad_pass) print(f"[LDAP] Подключено к ldap://172.16.20.20 как {ad_user}") if args.test_all: - test_all_methods(conn, ad_base) + test_all_methods(conn) else: - data = find_user(conn, ad_base, login=args.login, email=args.email, phone=args.phone) + data = find_user(conn, login=args.login, email=args.email, phone=args.phone) if data: print_report(data)