chore: remove test_ad.py from dispatcher branch
This commit is contained in:
-230
@@ -1,230 +0,0 @@
|
|||||||
#!/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())
|
|
||||||
Reference in New Issue
Block a user