diff --git a/config/__pycache__/config.cpython-311.pyc b/config/__pycache__/config.cpython-311.pyc new file mode 100644 index 0000000..e85e361 Binary files /dev/null and b/config/__pycache__/config.cpython-311.pyc differ diff --git a/test_ad.py b/test_ad.py new file mode 100644 index 0000000..3b4a14a --- /dev/null +++ b/test_ad.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +""" +Тестовый скрипт для проверки подключения к Active Directory. +Работает автономно - не требует Flask, Qdrant, Ollama, pip, dotenv. + +Использование: + python3 test_ad.py --login user123 + python3 test_ad.py --email user123@sibcem.ru + python3 test_ad.py --phone 7012345 +""" + +import os +import sys +import argparse +import re + + +# ==================== АВТОНОМНЫЙ ЗАГРУЗЧИК .env ==================== +def load_env_file(filepath): + """Парсит .env файл без зависимостей (аналог python-dotenv).""" + env = {} + if not os.path.exists(filepath): + return env + with open(filepath, 'r', encoding='utf-8') as f: + for line in f: + line = line.strip() + if not line or line.startswith('#'): + continue + if '=' in line: + key, value = line.split('=', 1) + key = key.strip() + value = value.strip().strip('"').strip("'") + env[key] = value + return env + + +# ==================== ЗАГРУЗКА КОНФИГА ==================== +BASE_DIR = os.path.abspath(os.path.dirname(__file__)) + +# Загружаем .env из config/ +env_file = os.path.join(BASE_DIR, 'config', '.env') +env_vars = load_env_file(env_file) + +# ==================== ПАРАМЕТРЫ AD ==================== +# Пути к Passwork (если есть) +PASSWORK_DIR = '/opt/passwork' +passwork_pool = {} +if os.path.exists(PASSWORK_DIR): + try: + sys.path.insert(0, PASSWORK_DIR) + from passwork import get_passwork_secrets + required = [env_vars.get('PW_ID_FTP', ''), env_vars.get('PW_ID_SD', ''), env_vars.get('PW_ID_AD', '')] + required = [r for r in required if r] + passwork_pool = get_passwork_secrets(required_cards=required) + except Exception: + passwork_pool = {} + + +def resolve_secret(pw_key, env_key): + """Получает секрет из Passwork или из .env.""" + card_name = env_vars.get(env_key, '').strip() + if card_name and card_name in passwork_pool: + return passwork_pool[card_name].get('password', '').strip() + return env_vars.get(env_key, '').strip() + + +# AD-параметры (дублируют config.py) +AD_SERVER = 'ldap://172.16.20.20' +AD_BASE = 'OU=-Пользователи,DC=sibcem,DC=ru' +AD_CONNECT_TIMEOUT = 5 +AD_RECEIVE_TIMEOUT = 5 +AD_USER = resolve_secret('AD', 'PW_ID_AD') or env_vars.get('AD_USER', '').strip() +AD_PASSWORD = resolve_secret('AD', 'PW_ID_AD') or env_vars.get('AD_PASSWORD', '').strip() + +# CITY_SUPPORT - встроенный справочник (дублирует config.py) +CITY_SUPPORT = { + 'Ангарск': 'Техподдержка Ангарск', + 'Слюдянка': 'Техподдержка Ангарск', + 'Искитим': 'Техподдержка Искитим', + 'Новосибирск': 'Техподдержка КТЦ', + 'Топки': 'Техподдержка Топки', + 'Кемерово': 'Техподдержка ХК', + 'Москва': 'Техподдержка Москва', + 'Тимлюй': 'Техподдержка ХК', + 'Каменск': 'Техподдержка ХК', + 'Улан-Удэ': 'Техподдержка ХК', + 'Татарский Ключ': 'Техподдержка ХК', + 'Спас-Клепики': 'Техподдержка ХК', + 'Красноярск': 'Техподдержка КЦ', +} + + +# ==================== ФУНКЦИИ ПОИСКА ==================== +def print_header(title): + print(f'\n{"="*60}\n {title}\n{"="*60}') + + +def print_result(label, value, success=True): + icon = '\u2705' if success else '\u274c' + print(f' {icon} {label}: {value}') + + +def search_ad(login=None, email=None, phone=None): + """Выполняет LDAP-поиск и возвращает данные пользователя.""" + if login: + ldap_filter = f'(sAMAccountName={login})' + search_by = f'Логину ({login})' + print_header(f'Поиск по логину: {login}') + elif email: + ldap_filter = f'(mail={email})' + search_by = f'Почте ({email})' + print_header(f'Поиск по email: {email}') + elif phone: + ldap_filter = f'(|(telephoneNumber=*{phone}*)(mobile=*{phone}*)(homePhone=*{phone}*)(pager=*{phone}*)(facsimileTelephoneNumber=*{phone}*)(ipPhone=*{phone}*))' + search_by = f'номеру телефона ({phone})' + print_header(f'Поиск по телефону: {phone}') + + try: + from ldap3 import Server, Connection, ALL + except ImportError: + print_result('Ошибка', 'Модуль ldap3 не установлен. Установите: pip install ldap3', False) + return None + + try: + server = Server(AD_SERVER, get_info=ALL, connect_timeout=AD_CONNECT_TIMEOUT) + conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True, receive_timeout=AD_RECEIVE_TIMEOUT) + conn.search(search_base=AD_BASE, search_filter=ldap_filter, + attributes=['l', 'company', 'manager', 'mail', 'displayName', 'telephoneNumber']) + except Exception as e: + print_result('Ошибка подключения к AD', str(e), False) + return None + + if not conn.entries: + print_result('Результат', 'Пользователь не найден', False) + return None + + user = conn.entries[0] + + # Парсим атрибуты + city = user.l.value if 'l' in user and user.l.value else None + company = user.company.value if 'company' in user and user.company.value else None + ad_path = user.entry_dn + display_name = user.displayName.value if 'displayName' in user and user.displayName.value else None + mail = user.mail.value if 'mail' in user and user.mail.value else None + phone_val = user.telephoneNumber.value if 'telephoneNumber' in user and user.telephoneNumber.value else None + + # Город из OU + if not city and ad_path: + match = re.search(r'OU=([^,]+),OU=-Пользователи', ad_path, re.IGNORECASE) + if match: + city = match.group(1).strip() + if city.lower() == 'красноярск' and 'OU=ООО Комбинат Волна' in ad_path: + company = 'ООО Комбинат Волна' + + # Email руководителя + manager_dn = user.manager.value if 'manager' in user else None + manager_email = None + if manager_dn: + try: + server2 = Server(AD_SERVER, get_info=ALL, connect_timeout=AD_CONNECT_TIMEOUT) + conn2 = Connection(server2, user=AD_USER, password=AD_PASSWORD, auto_bind=True, receive_timeout=AD_RECEIVE_TIMEOUT) + conn2.search(search_base=manager_dn, search_filter='(objectClass=user)', attributes=['mail'], search_scope='BASE') + if conn2.entries and 'mail' in conn2.entries[0]: + manager_email = conn2.entries[0].mail.value + except Exception: + pass + + # Вывод результатов + print_result('ФИО', display_name or 'Не указано') + print_result('Email', mail or 'Не указано') + print_result('Телефон', phone_val or 'Не указано') + print_result('Город', city or 'Не определено', city is not None) + print_result('Компания', company or 'Не определена', company is not None) + print_result('OU-путь', ad_path) + print_result('Руководитель', manager_email or 'Не указан', manager_email is not None) + print_result('Источник', f'AD ({search_by})') + + # Региональная техподдержка + if city: + target_group = CITY_SUPPORT.get(city.strip()) + if target_group: + print_result('\U0001f4cd Региональная техподдержка', target_group) + elif city.strip() == 'Красноярск': + print_result('\U0001f4cd Региональная техподдержка', 'Техподдержка КЦ (по умолчанию)') + + return { + 'city': city, 'company': company, 'ad_path': ad_path, + 'manager_email': manager_email, 'display_name': display_name, + 'mail': mail, 'phone': phone_val, 'source': f'AD ({search_by})', + } + + +def print_config_info(): + """Вывод конфигурации.""" + print_header('Конфигурация AD') + print_result('AD Server', AD_SERVER) + print_result('Base DN', AD_BASE) + print_result('Connect Timeout', f'{AD_CONNECT_TIMEOUT} сек') + print_result('Receive Timeout', f'{AD_RECEIVE_TIMEOUT} сек') + print_result('Bind User', AD_USER or 'Не задан') + print_result('.env файл', env_file) + print_result('Справочник городов', f'{len(CITY_SUPPORT)} городов') + + +def main(): + parser = argparse.ArgumentParser(description='Тест подключения к Active Directory') + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument('--login', '-l', help='sAMAccountName пользователя') + group.add_argument('--email', '-e', help='Email пользователя') + group.add_argument('--phone', '-p', help='Номер телефона') + args = parser.parse_args() + + print_header('SD Dispatcher - Тест подключения к AD') + print(f' Проект: {BASE_DIR}') + print(f' Config: {os.path.join(BASE_DIR, "config", "config.py")}') + print() + print_config_info() + + result = search_ad(login=args.login, email=args.email, phone=args.phone) + + if result: + print(f'\n{"="*60}\n \u2705 Подключение к AD работает!\n{"="*60}') + return 0 + else: + print(f'\n{"="*60}\n \u274c Не удалось получить данные из AD\n{"="*60}') + return 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/test_ad_lookup.py b/test_ad_lookup.py new file mode 100644 index 0000000..f078d75 --- /dev/null +++ b/test_ad_lookup.py @@ -0,0 +1,283 @@ +#!/usr/bin/env python3 +""" +Тестовый скрипт для проверки обращения к Active Directory. +Аналог find_user_in_ad() из sd_dispatcher.py — standalone. + +Установка зависимостей (на сервере 192.168.1.106): + python3 -m pip install ldap3 + +Использование: + # 1) Через аргументы: + python3 test_ad_lookup.py --user "sibcem\\svc_dispatcher" --pass "password" --login ivanov + + # 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 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" + + return user, password, base + + +# --------------------------------------------------------------------------- +# 2. Подключение к LDAP +# --------------------------------------------------------------------------- + +def connect_to_ad(ad_user, ad_pass, ad_base): + """Возвращает подключённый ldap3.Connection.""" + from ldap3 import Server, Connection + + server = Server( + "ldap://172.16.20.20", + get_info=None, # не тянем schema — экономим время + connect_timeout=5, + ) + conn = Connection( + server, + user=ad_user, + password=ad_pass, + auto_bind=True, + receive_timeout=5, + ) + return conn, ad_base + + +# --------------------------------------------------------------------------- +# 3. Поиск пользователя +# --------------------------------------------------------------------------- + +def find_user(conn, ad_base, login=None, email=None, phone=None): + """ + Ищет пользователя в AD. Приоритет: телефон > логин > email. + Возвращает dict с атрибутами или None. + """ + if phone: + filt = ( + f"(|(telephoneNumber=*{phone}*)(mobile=*{phone}*) " + f"(homePhone=*{phone}*)(pager=*{phone}*) " + f"(facsimileTelephoneNumber=*{phone}*)(ipPhone=*{phone}*))" + ) + by = f"телефону {phone}" + elif login: + clean = login.split("\\")[-1].split("/")[-1] + filt = f"(sAMAccountName={clean})" + by = f"логину {clean}" + elif email: + filt = f"(mail={email})" + by = f"почте {email}" + else: + print("❌ Укажите --login, --email или --phone") + sys.exit(1) + + print(f"\n[LDAP] Поиск: {filt}") + print(f"[LDAP] Base DN: {ad_base}") + + # Запрос 1 — основной + conn.search( + search_base=ad_base, + search_filter=filt, + attributes=["l", "company", "manager", "department", "ou", + "displayName", "mail", "sAMAccountName", "title", + "telephoneNumber", "mobile", "givenName", "sn"], + ) + + if not conn.entries: + print(f"\n❌ Пользователь не найден по {by}") + return None + + user = conn.entries[0] + print(f"\n✅ Найдено: {user.entry_dn}") + + result = { + "sam": user.sAMAccountName.value if "sAMAccountName" in user else None, + "given_name": user.givenName.value if "givenName" in user else None, + "surname": user.sn.value if "sn" in user else None, + "display_name": user.displayName.value if "displayName" in user else None, + "mail": user.mail.value if "mail" in user else None, + "title": user.title.value if "title" in user else None, + "department": user.department.value if "department" in user else None, + "city": user.l.value if "l" in user and user.l.value else None, + "company": user.company.value if "company" in user and user.company.value else None, + "dn": user.entry_dn, + "phone": user.telephoneNumber.value if "telephoneNumber" in user else None, + "mobile": user.mobile.value if "mobile" in user else None, + } + + # Город из 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']}") + + # Запрос 2 — email руководителя + manager_dn = user.manager.value if "manager" in user else None + if manager_dn: + conn.search( + search_base=manager_dn, + search_filter="(objectClass=user)", + attributes=["mail", "displayName"], + search_scope="BASE", + ) + 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 + else: + result["manager_email"] = None + else: + result["manager_email"] = None + + return result + + +# --------------------------------------------------------------------------- +# 4. Форматирование вывода +# --------------------------------------------------------------------------- + +def print_report(data): + """Красивый вывод результатов.""" + print("\n" + "=" * 60) + print(" РЕЗУЛЬТАТ ПОИСКА В ACTIVE DIRECTORY") + print("=" * 60) + print(f" sAMAccountName: {data['sam']}") + print(f" Имя: {data['given_name']}") + print(f" Фамилия: {data['surname']}") + print(f" ФИО: {data['display_name']}") + print(f" Email: {data['mail']}") + print(f" Должность: {data['title']}") + print(f" Отдел: {data['department']}") + print(f" Город: {data['city'] or '—'}") + print(f" Компания: {data['company'] or '—'}") + print(f" Телефон: {data['phone'] or '—'}") + print(f" Мобильный: {data['mobile'] or '—'}") + print(f" Руководитель: {data['manager_name'] or '—'}") + if data.get("manager_email"): + print(f" Email руково.: {data['manager_email']}") + print("-" * 60) + 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=")] + if ou_parts: + print(f"\n OU-структура:") + for i, ou in enumerate(reversed(ou_parts)): + indent = " " * i + print(f" {indent}└── {ou}") + + print("=" * 60) + + +# --------------------------------------------------------------------------- +# 5. Тестирование всех методов поиска +# --------------------------------------------------------------------------- + +def test_all_methods(conn, ad_base): + """Находит первого пользователя и проверяет все методы поиска.""" + print("\n" + "─" * 60) + print(" РЕЖИМ ТЕСТИРОВАНИЯ: поиск по всем методам") + print("─" * 60) + + conn.search( + search_base=ad_base, + search_filter="(objectClass=user)", + attributes=["sAMAccountName", "mail", "telephoneNumber", "displayName", + "l", "company", "department"], + size_limit=1, + ) + + if not conn.entries: + print("❌ В AD нет пользователей.") + return + + sample = conn.entries[0] + sam = sample.sAMAccountName.value + 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})") + + 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) + if r: + print_report(r) + + print("\n" + "─" * 60) + 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)") + 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="Номер телефона для поиска") + parser.add_argument("--test-all", action="store_true", + 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") + + 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) + print(f"[LDAP] Подключено к ldap://172.16.20.20 как {ad_user}") + + if args.test_all: + test_all_methods(conn, ad_base) + else: + data = find_user(conn, ad_base, login=args.login, email=args.email, phone=args.phone) + if data: + print_report(data) + + conn.unbind()