Denis Krivchenko
231 lines
9.8 KiB
Python
231 lines
9.8 KiB
Python
#!/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())
|