#!/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())