From 33a480a0cd0dccf7dd4684ca4ebb10d46f313bab Mon Sep 17 00:00:00 2001 From: Denis Krivchenko Date: Sat, 4 Jul 2026 08:24:34 +0700 Subject: [PATCH] =?UTF-8?q?add:=20=D1=82=D0=B5=D1=81=D1=82=D0=BE=D0=B2?= =?UTF-8?q?=D1=8B=D0=B9=20=D1=81=D0=BA=D1=80=D0=B8=D0=BF=D1=82=20test=5Fad?= =?UTF-8?q?.py=20=D0=B8=20INSTALL.md=20=D0=B4=D0=BB=D1=8F=20=D0=BF=D0=B5?= =?UTF-8?q?=D1=80=D0=B5=D0=BD=D0=BE=D1=81=D0=B0=20=D0=BD=D0=B0=20=D0=BD?= =?UTF-8?q?=D0=BE=D0=B2=D1=8B=D0=B9=20=D1=81=D0=B5=D1=80=D0=B2=D0=B5=D1=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/__pycache__/config.cpython-311.pyc | Bin 0 -> 8877 bytes test_ad.py | 230 ++++++++++++++++++ test_ad_lookup.py | 283 ++++++++++++++++++++++ 3 files changed, 513 insertions(+) create mode 100644 config/__pycache__/config.cpython-311.pyc create mode 100644 test_ad.py create mode 100644 test_ad_lookup.py diff --git a/config/__pycache__/config.cpython-311.pyc b/config/__pycache__/config.cpython-311.pyc new file mode 100644 index 0000000000000000000000000000000000000000..e85e361825b7572573d0a1409bfaec0db050fc93 GIT binary patch literal 8877 zcmcgQYj9IndRMk=%a&}*mMj@R#UlX&eqv)|2tu~5ZDGkWlFZ{mqR2iMBd{&!N;04& zGa}12Buk5v(g|fD7&1#}ciT7*Hb4l;OuPNn&Ya6=hCBCHy6tqzw6pb2%FIrF_4|$_ zKOjSBrqe6wyXSo8JKy_!=Nx@lTwKJ#^OkYo=->5o+`nKa`;=rF-~PUUG?h zkbY|f4fJy+awcLQrDr(NFlWm0S^mBOonI{?oZQO3UE}^HaYB_)EmlX3jGI}k`UOWSuH-(+^<1S`{d`#|)QA;xwaX=E z+hWa1u0owym#k-UEn*GGwLUEuMxW^&v5r(OnfrVi0?+!D-Vm#HfRbVrS#^x9L8(Km z`+gfp)sM^Rr)83AlsVN6Qfo3&+y5V=)}mDA^EwvmdggiCmAsX)(LjEjD<`W38(AZ) zCToQ?WSy|~xvJj_D_5Yfj=U(mK-LQ{k`2Oo(jaVjE-o9N;L`A1vo<|3Ya`h#Y$A=q zX3``yf*wuqYbMP?i`WAB)^a6BXeF&e8)*}^kS#(xX%{-+x0Q4VoiMWt#<}E}C83Ax6ncqWaFSgDPaHxYNa`m&!T{+NT=3pa zoWdaC1vlvvJfvS3A_IaK=!c0**h6*;J~AlyiCYK|k1#@pguTQo>?6a%e!v|ddjx^_ zgoDH{90IzRLC06dITE<85rXg@f%mIqFWEP)32TH9*$g8QE^NXW6`(>in_i1-L{3R7iMkr zefqJSr1SDy^cMY+-ln%~^sbHGqxb1&@=xg(@}Jp;GaPO7wvB!w|D1j*Ct>JwdY8`I zkXdrI?8 z&bb0V_g9>R+g?MCn$y&xYNtYzVifC@&Gy||W?B^d8jenfVq-@{sX6W1%&Teo!NlM^DPi?;H(l!d@UZ9L=qeeNhBH% z1NwJY&e`ZCI!7+t)K{sp}T{aN}E`t2s5-ob9Sja~u2flof8^YjaPUp_6v#IMkY z;3V*&{3g9Ezk$dl$DUhsY`5(Zx+)$8uq-lj5S7#PD~Qf5;ByNYXCuU%eisMiH$ccI zEHWUMehLBnGK~T0%cY&C_b@=Wkt0q6jc#GwAx`lgH1*Qlf89QBqwlgLxCeHyWMHPv zL##i^Vjw+0uOtcv9(OS((s7R;05&$51{w4@P@^wE7f2Z!W`C27USz`WgG0{I&%s-$ z#%ahHbQyAG1E@aZ@Fj==CDVE|#RpNz=Z?pv6*^*?f{(!Y$nF-nVtu?3_^r>V3!>;9 z=3{7~!py;#)wdRdv|80=mJ*ThWUS^*M8?BY>7_d2=Y0~MYGJ)PgV>A19RM=d_Pfh} zZj0~5HhO9?MvyF-NStL(f(}GZ{uzYiKC}392o<)O-{qg9_MboP`t{#w)+cnWElv38 zOcXRHLZhSM6N%DHqd6pv9SP5f;j^$H66H&)CkV5{-qwgk!5(5DSkTgR=R^jO?n>I@GMKlexSV z3~Fs@7AHqVk}yxkCt~4=@YHxbG|?z_G>(R%Vxs1IrgXJ}=~8IwSfXZhTx?9YMX7P7 zt#NWo+T%iHane1<2P6|9^s&%k=u=dYpWfaf7WA zrrd)MR9+}|u#hi-_#5#Kq`8*HVnxC#fOQY+CJj2sFe01I*B0WS2ed^(Q%D@gqtdco#|y>2WPpNzL5ZuW=7q5!55 zDr@U;#dT>`@4zN_jqM=7_X{ZMzX1n94aI792E3aDYv$!Yfj)Wz_$E3rc>< zwv^NK?hnvj1e0J}M1$zP$7`R-_d0U|P^WX|8P;OXJe$|M!0QXpbN=zOt-g%z1`T0n zk#G}~HUdbP#ip5vI1?e^sikTt!J6l!Rs>rRV5&&X2-*=e0ca?G%<^o7hvG&UfiSX# zBA*APaG^4~mEOU_Ae^0Wn?t2;LuSCXhw43M%XP^=jN>J?QgG?B8IKJ(eVILRhFBg} zCKDGo+xn4ldJTL4u7cx9Bdpuc(@&yju+Ocr#q*{QxM`-G#WJeC$5Nt<)wMyznuoVH;#s<1fC!=b9`jRAJ0avwftUsa_kg$a3ezhPp ziU<88sxBBj9G(z^K~>k|@bf{Z%LfOP2pqif;dXsiWKN_YSdEaqqy$&|uKd_j-q%eznl)>J7jRH87wW zGsnUuj5WjC$M^G~g4gHcdl~2KK;MWU99TY31ZU{Iye~N7bE`Uj$T{qF4F%LPC*S87 zaR-7ve$NQ+hjS#qb`7%N^|Vc|B@{!_E5welXpSE>55F{lGSjcDUWZws(*Z_*MNLr_V7INCz_8 zVT{W%BfVa?JL3UU2Ic25$U(Qu;{rQZ-tW-~<|UNJ>*U?RA%_PN!pDQC!HlC!i@jwe zmL(*F-|6kmq=0b|pLJ2`B4!`&>K_QG#hI{ywhoU&ElR&xObnSe6Qq`Bdj64~-RN|t z@DsQ+WvM>^*#+@(_416YBtted;tBQ!_Ajyu@Sb6?EimHa)v~NGFo6Bn;a9WMvbvig z-hD180Khgd;N$%Ra5YhD`n*0sWHo`rTRvKqaXqeqz1{%t4}!#Czt1}|tX9LC0Yf~F z00hPV#Kf9@hp)%c&tsB3Ewnm2Ix^&fEKM`?XChmb9o!4c%dt}1`s|p;HI$tN1_ZOD zYGsz7=K$6<2qsV>6V;kA@u`8Dl>^9mNoln@GwSH=<%a`Ls8hPg9-KZL98}F%FtEv7 zl(opP*WreLwbD3s6)XhrK3E>0pR0e!0~I+tRKiHWr9FC%$^cYB_P>&Z8?{sp z;5*%p$q0#0h&!c^p^N{$L_dM+#zL+}qfxlRB>N~_QIdTWu8yw!F648Dl5?Hs?Qh#J zwq5SLYQJQ^+4k$skL-VKzuhHoa;7#pp9Z+wO~% zb6q$0%7*rop4rP-drP_8dIjmho&tLOk3u6%O-ouWLHcz zY*@HiE}J%|Oq(B?+8&tN=DTIn&Xj5A4-799XiYf_9Dro$LJ4QJ&5m9OQ(MQpBwM;u zmhOj^T@Ng~zOIrj?v%w%ITPq(tWO%0+|t?Tg%dQl9=?jU;LO0A-Z#9nj@eVPc3n!l zj%wE}XbUw)McYVMeidzDlBY$l$yzC;m8ez%FH{a{;Gegf^$#2D4;t)pgCo`80EBGa zor1sUv}VP;fx@?IfSO&h*_AT8lAfgJk4k}Yfy>b>0~zKg>CACjbP~R@Hj&aMs5YT! z4awd!184V~9y~LcT=w}RJPUci1>ZVOlb_6^`9G1lQz`Bg$c-#}%H)`ELdR4{b$f z{KiuMUgX`#+)>%mnzFPey^68+p>g#CA;$g?;-L-f&*q-f&6d5TKc_#d2R~P@o7{-|8#NL4wK{mK5* zt~0K)N5CK;(ivyF9~9Kmf?CDa@L}=w;+x};4Yux-t^1*E*8>}7gUy|?xz84yEqJT& zT;c3SSyxMSwMt>x?8^@d*U-W>O8thbuU~ro=Dzt+xqe%!e%r%(`-6J>*F|#uV5)v_ zab|<8tN8{K0yNh5=gbbI--|G=aiTFWv>bw|c>`!FK)?e`V<3D6OH!v4nUV!cNqthk zVBn1AqydaFTP}3Ix8vO%b2_?lXL{jK&WHn*>G!G{%2hzW!b-U^SP8XjD7Omw#ifgj zvr{%~N*OjK3l)7aEb|?2@3^R&dr{Umr1TAJF_)Q=#W-aRkeFK1e5>mzXH}}8R#<^+ zNn5f=DXFCTjp;9&R;Xhmy=t_tW%$^n>bk!L<RMm=RoP zvanK6R>6!Z*#;Vyu47KLI5eo0wF$Z8bEXa;%;(Cj$r5;3KvKDaOv)6PTLNM#!KY^6 zXakjNfTYewxhitp!9DVDyY!UOipiNo`#VxjE%Ipp${K?$Rig4#i?u>g-B+r&8Ie zn71qDEsCW>v2In?_9*o`m747f)g{Ot0B~NDwnQ;)qUD=Gf%0C;zRA*WimPbN2sP|g z45eo~7=VguMLn!J=w2gNv59sbr20efRSf0S{36w_XN}EN-^dz}3!7MD9o4_^?8fX6 y)vsMnGV7-L)eBZbIfw+1EWu9{{~7!Hj_rIa_uJOIzPjAsZmsKEmHQ8?u>C(^(D}Cj literal 0 HcmV?d00001 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()