AD: поиск по нескольким OU (AD_BASES)
- config.py: AD_BASE -> AD_BASES (кортеж OU) - utils/ad_search.py: новая утилита для поиска по всем OU - utils/ad_checker.py: использовать ad_search - utils/otp_service.py: использовать ad_search - lk/handlers.py: использовать ad_search - transcription_bot/handlers.py: использовать ad_search - service_desk/handlers.py: использовать ad_search - instruct/handlers.py: использовать ad_search - photo_bot/ad_helper.py: использовать ad_search - photo_bot/check_ad_photo.py: использовать ad_search - check_ad_2fa.py: использовать ad_search - debug_ad_filter.py: обновить для AD_BASES - lk/test_lk_payslip.py: обновить для AD_BASES
This commit is contained in:
+10
-11
@@ -13,27 +13,26 @@ from ldap3 import Server, Connection, ALL, SUBTREE
|
||||
|
||||
# Подтягиваем конфиг
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE
|
||||
from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASES
|
||||
from utils.ad_search import search_by_filter as ad_search_all_bases
|
||||
|
||||
|
||||
def find_user(login: str) -> dict | None:
|
||||
"""Найти пользователя в AD по логину (sAMAccountName или mail)."""
|
||||
server = Server(AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True)
|
||||
|
||||
"""Найти пользователя в AD по логину (sAMAccountName или mail).
|
||||
Ищет по всем OU из AD_BASES."""
|
||||
filter_login = login.lower().replace("\\", "\\5c").replace("*", "\\2a").replace("(", "\\28").replace(")", "\\29")
|
||||
|
||||
if "@" in filter_login:
|
||||
ldap_filter = f"(|(mail={filter_login})(sAMAccountName={filter_login}))"
|
||||
ldap_filter = "(|{}{})".format("(mail={})".format(filter_login), "(sAMAccountName={})".format(filter_login))
|
||||
else:
|
||||
ldap_filter = f"(|(sAMAccountName={filter_login})(mail={filter_login}))"
|
||||
ldap_filter = "(|{}{})".format("(sAMAccountName={})".format(filter_login), "(mail={})".format(filter_login))
|
||||
|
||||
conn.search(search_base=AD_BASE, search_filter=ldap_filter, attributes=["cn", "sAMAccountName", "memberOf"])
|
||||
entries = ad_search_all_bases(ldap_filter, ["cn", "sAMAccountName", "memberOf"])
|
||||
|
||||
if not conn.entries:
|
||||
if not entries:
|
||||
return None
|
||||
|
||||
entry = conn.entries[0]
|
||||
entry = entries[0]
|
||||
return {
|
||||
"dn": entry.entry_dn,
|
||||
"cn": str(entry.cn),
|
||||
@@ -64,7 +63,7 @@ def main():
|
||||
print(" Проверка пользователя в группе 2FA (Active Directory)")
|
||||
print("=" * 60)
|
||||
print(f"\n Сервер AD: {AD_SERVER}")
|
||||
print(f" База поиска: {AD_BASE}")
|
||||
print(f" Базы поиска: {config.AD_BASES}")
|
||||
print(f" Подключено как: {AD_USER}")
|
||||
print()
|
||||
|
||||
|
||||
+4
-1
@@ -104,7 +104,10 @@ ALLOWED_USERS = [ # Список TrueConf ID разрешенны
|
||||
TC_SERVER = os.getenv("TC_SERVER", "") # Адрес сервера TrueConf
|
||||
|
||||
AD_SERVER = "ldap://172.16.20.20" # Адрес контроллера домена Active Directory
|
||||
AD_BASE = "OU=-Пользователи,DC=sibcem,DC=ru" # Базовый путь поиска пользователей в AD
|
||||
AD_BASES = ( # Базовые пути поиска пользователей в AD (несколько OU)
|
||||
"OU=-Пользователи,DC=sibcem,DC=ru",
|
||||
"OU=Планшеты,OU=enabled,OU=БезКомпьютеров,DC=sibcem,DC=ru",
|
||||
)
|
||||
|
||||
SQL_SERVER = "SRVKEM-MOBILEIN.sibcem.ru" # Сервер MS SQL для работы Личного кабинета
|
||||
SQL_DB_NAME = "BossCopy" # База данных с информацией по кадрам и ЗП
|
||||
|
||||
+19
-15
@@ -3,7 +3,8 @@
|
||||
|
||||
import sys
|
||||
sys.path.insert(0, "/opt/trueconf_bot")
|
||||
from config.config import AD_USER, AD_PASSWORD, AD_SERVER, AD_BASE
|
||||
from config.config import AD_USER, AD_PASSWORD, AD_SERVER, AD_BASES
|
||||
from utils.ad_search import search_by_filter as ad_search_all_bases
|
||||
|
||||
from ldap3.utils.conv import escape_filter_chars
|
||||
|
||||
@@ -19,7 +20,14 @@ print(f"short_username: {short_username}")
|
||||
print(f"safe_user_id: {safe_user_id}")
|
||||
|
||||
# lk/handlers.py filter
|
||||
filter_lk = f"(&(objectClass=user)(|(mail={safe_user_id})(userPrincipalName={safe_user_id})(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')}))(sAMAccountName={escape_filter_chars(short_username)})))"
|
||||
filter_lk = "(&(objectClass=user)(|(mail={})(userPrincipalName={})(userPrincipalName={}{}{}))({}))".format(
|
||||
safe_user_id,
|
||||
safe_user_id,
|
||||
escape_filter_chars(short_username),
|
||||
"@sibcem.ru",
|
||||
"",
|
||||
escape_filter_chars(short_username),
|
||||
)
|
||||
print(f"\nFilter (lk format):")
|
||||
print(filter_lk)
|
||||
|
||||
@@ -29,27 +37,23 @@ closes = filter_lk.count(')')
|
||||
print(f"\nOpen parens: {opens}, Close parens: {closes}")
|
||||
|
||||
# Now test with the connection
|
||||
from ldap3 import Server, Connection, ALL
|
||||
server = Server(AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True)
|
||||
print(f"\nSearching with filter across all AD_BASES...")
|
||||
entries = ad_search_all_bases(filter_lk, ["cn", "sAMAccountName", "mail", "userPrincipalName"])
|
||||
|
||||
print(f"\nSearching with filter...")
|
||||
conn.search(AD_BASE, filter_lk, attributes=["cn", "sAMAccountName", "mail", "userPrincipalName"])
|
||||
|
||||
if conn.entries:
|
||||
entry = conn.entries[0]
|
||||
if entries:
|
||||
entry = entries[0]
|
||||
print(f"FOUND: cn={entry.cn}, mail={entry.mail if 'mail' in entry else 'N/A'}, sam={entry.sAMAccountName if 'sAMAccountName' in entry else 'N/A'}")
|
||||
else:
|
||||
print("NOT FOUND in AD_BASE")
|
||||
print("NOT FOUND in any AD_BASE")
|
||||
|
||||
# Try root
|
||||
conn2 = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True)
|
||||
root_base = "DC=sibcem,DC=ru"
|
||||
print(f"\nSearching in {root_base}...")
|
||||
conn2.search(root_base, filter_lk, attributes=["cn", "sAMAccountName", "mail", "userPrincipalName"])
|
||||
from utils.ad_search import search_by_filter
|
||||
entries2 = search_by_filter(filter_lk, ["cn", "sAMAccountName", "mail", "userPrincipalName"])
|
||||
|
||||
if conn2.entries:
|
||||
entry = conn2.entries[0]
|
||||
if entries2:
|
||||
entry = entries2[0]
|
||||
print(f"FOUND: cn={entry.cn}, mail={entry.mail if 'mail' in entry else 'N/A'}, sam={entry.sAMAccountName if 'sAMAccountName' in entry else 'N/A'}")
|
||||
else:
|
||||
print("NOT FOUND in root")
|
||||
|
||||
+7
-24
@@ -21,15 +21,14 @@ logger = logging.getLogger(__name__)
|
||||
router = Router()
|
||||
|
||||
|
||||
def _get_user_cn(user_id: str) -> str:
|
||||
def get_user_cn(user_id: str) -> str | None:
|
||||
"""
|
||||
Получить CN пользователя из AD по его user_id (email или логин).
|
||||
user_id из TrueConf — это обычно email, например 'ds.krivochenko@tcs.sibcem.ru'
|
||||
Заменяет tcs.sibcem.ru на sibcem.ru для поиска в AD.
|
||||
Ищет по всем OU из AD_BASES.
|
||||
"""
|
||||
from ldap3 import Server, Connection, ALL
|
||||
from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE
|
||||
from ldap3.utils.conv import escape_filter_chars
|
||||
from utils.ad_search import search_by_user_id
|
||||
|
||||
try:
|
||||
# Заменяем tcs.sibcem.ru на sibcem.ru
|
||||
@@ -38,26 +37,10 @@ def _get_user_cn(user_id: str) -> str:
|
||||
|
||||
logger.info(f"AD lookup: user_id={user_id}, search_id={search_id}, short={short_username}")
|
||||
|
||||
server = Server(AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True)
|
||||
entries = search_by_user_id(search_id, ["cn", "sAMAccountName", "mail", "userPrincipalName"])
|
||||
|
||||
# escape_filter_chars экранирует @ и . в email, что ломает фильтр.
|
||||
# Для email и UPN передаём raw. Для sAMAccountName — экранируем.
|
||||
safe_sam = escape_filter_chars(short_username)
|
||||
|
||||
search_filter = (
|
||||
f"(&(objectClass=user)(|"
|
||||
f"(mail={search_id})"
|
||||
f"(userPrincipalName={search_id})"
|
||||
f"(userPrincipalName={short_username}@sibcem.ru))"
|
||||
f"(sAMAccountName={safe_sam}))"
|
||||
)
|
||||
logger.info(f"AD filter: {search_filter}")
|
||||
|
||||
conn.search(AD_BASE, search_filter, attributes=["cn", "sAMAccountName", "mail", "userPrincipalName"])
|
||||
|
||||
if conn.entries:
|
||||
entry = conn.entries[0]
|
||||
if entries:
|
||||
entry = entries[0]
|
||||
cn = str(entry.cn)
|
||||
mail = str(entry.mail) if 'mail' in entry else "N/A"
|
||||
sam = str(entry.sAMAccountName) if 'sAMAccountName' in entry else "N/A"
|
||||
@@ -110,7 +93,7 @@ async def instruct_router_handler(msg: Message):
|
||||
elif cmd == "1":
|
||||
log_menu_stats(user_id, "Инструкции", "Trueconf")
|
||||
set_state(user_id, "INSTRUCT_TRUECONF_VIEW")
|
||||
cn = _get_user_cn(user_id)
|
||||
cn = get_user_cn(user_id)
|
||||
text = INSTRUCT_TRUECONF_WITH_AD(EMOJI_DIGITS, cn)
|
||||
await msg.answer(text, parse_mode="html")
|
||||
elif cmd == "2":
|
||||
|
||||
+10
-11
@@ -129,19 +129,18 @@ async def handle_lk_error(msg: Message, user_id: str, action: str, error_details
|
||||
# ИНТЕГРАЦИЯ С ACTIVE DIRECTORY ЧЕРЕЗ extensionAttribute2
|
||||
# =========================================================
|
||||
def get_card_id_from_ad(user_id: str) -> str:
|
||||
# Поиск CARD_ID по всем OU из AD_BASES
|
||||
from utils.ad_search import search_by_user_id
|
||||
try:
|
||||
server = Server(config.AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=config.AD_USER, password=config.AD_PASSWORD, auto_bind=True)
|
||||
safe_user_id = escape_filter_chars(user_id)
|
||||
short_username = user_id.split('@')[0]
|
||||
search_filter = f"(&(objectClass=user)(|(mail={safe_user_id})(userPrincipalName={safe_user_id})(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')})(sAMAccountName={escape_filter_chars(short_username)})))"
|
||||
|
||||
conn.search(config.AD_BASE, search_filter, attributes=['extensionAttribute2'])
|
||||
|
||||
if conn.entries and 'extensionAttribute2' in conn.entries[0] and conn.entries[0].extensionAttribute2.value:
|
||||
return str(conn.entries[0].extensionAttribute2.value).strip()
|
||||
entries = search_by_user_id(user_id, ['extensionAttribute2'])
|
||||
if not entries:
|
||||
raise Exception("Пользователь не найден в AD.")
|
||||
entry = entries[0]
|
||||
if 'extensionAttribute2' in entry and entry.extensionAttribute2.value:
|
||||
return str(entry.extensionAttribute2.value).strip()
|
||||
raise Exception("Поле extensionAttribute2 не заполнено в AD.")
|
||||
except Exception as e: raise Exception(f"AD Card Query Error: {e}")
|
||||
except Exception as e:
|
||||
raise Exception("AD Card Query Error: {}".format(e))
|
||||
|
||||
# =========================================================
|
||||
# МОДНОЕ УНИВЕРСАЛЬНОЕ ЯДРО ПАГИНАЦИИ СПИСКОВ ДЛЯ ЧАТА
|
||||
|
||||
+22
-13
@@ -13,7 +13,10 @@ from ldap3.utils.conv import escape_filter_chars
|
||||
TEST_USER = "man.bogov@tcs.sibcem.ru"
|
||||
|
||||
AD_SERVER = "ldap://172.16.20.20"
|
||||
AD_BASE = "OU=-Пользователи,DC=sibcem,DC=ru"
|
||||
AD_BASES = (
|
||||
"OU=-Пользователи,DC=sibcem,DC=ru",
|
||||
"OU=Планшеты,OU=enabled,OU=БезКомпьютеров,DC=sibcem,DC=ru",
|
||||
)
|
||||
|
||||
AD_USER = "sdesk-mail"
|
||||
AD_PASSWORD = "fne?e!q.m8phcrGVAcqr"
|
||||
@@ -32,7 +35,6 @@ DB_PASSWORD = "Bav:fX#UwH8%atv4"
|
||||
def get_phone_from_ad(user_id: str) -> str:
|
||||
|
||||
server = Server(AD_SERVER, get_info=ALL)
|
||||
|
||||
conn = Connection(
|
||||
server,
|
||||
user=AD_USER,
|
||||
@@ -41,22 +43,29 @@ def get_phone_from_ad(user_id: str) -> str:
|
||||
)
|
||||
|
||||
safe_user_id = escape_filter_chars(user_id)
|
||||
|
||||
short_username = user_id.split("@")[0]
|
||||
|
||||
search_filter = (
|
||||
f"(&(objectClass=user)"
|
||||
f"(|(mail={safe_user_id})"
|
||||
f"(userPrincipalName={safe_user_id})"
|
||||
f"(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')})"
|
||||
f"(sAMAccountName={escape_filter_chars(short_username)})))"
|
||||
"(&(objectClass=user)"
|
||||
"(|(mail={})"
|
||||
"(userPrincipalName={})"
|
||||
"(userPrincipalName={}))"
|
||||
"(sAMAccountName={}))".format(
|
||||
safe_user_id,
|
||||
safe_user_id,
|
||||
escape_filter_chars(short_username) + "@sibcem.ru",
|
||||
escape_filter_chars(short_username),
|
||||
)
|
||||
)
|
||||
|
||||
conn.search(
|
||||
AD_BASE,
|
||||
search_filter,
|
||||
attributes=['mobile', 'telephoneNumber']
|
||||
)
|
||||
for base in AD_BASES:
|
||||
conn.search(
|
||||
base,
|
||||
search_filter,
|
||||
attributes=['mobile', 'telephoneNumber']
|
||||
)
|
||||
if conn.entries:
|
||||
break
|
||||
|
||||
if not conn.entries:
|
||||
raise Exception("Пользователь не найден в AD")
|
||||
|
||||
+15
-13
@@ -5,37 +5,39 @@ from ldap3 import Server, Connection, ALL, MODIFY_REPLACE
|
||||
|
||||
# Подтягиваем конфиг
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE
|
||||
from config.config import AD_SERVER, AD_USER, AD_PASSWORD
|
||||
from utils.ad_search import search_by_login as ad_search_all_bases
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def upload_photo_to_ad(target_user: str, photo_path: str) -> dict:
|
||||
"""
|
||||
Читает готовое фото с диска и записывает его в атрибут thumbnailPhoto в AD.
|
||||
Ищет пользователя по всем OU из AD_BASES.
|
||||
"""
|
||||
from utils.ad_search import search_by_filter
|
||||
try:
|
||||
# Читаем сырые байты картинки
|
||||
with open(photo_path, "rb") as f:
|
||||
photo_bytes = f.read()
|
||||
|
||||
# Формируем фильтр для поиска
|
||||
# Ищем пользователя в AD по всем OU
|
||||
if '@' in target_user:
|
||||
ldap_filter = f"(mail={target_user})"
|
||||
ldap_filter = "(mail={})".format(target_user)
|
||||
else:
|
||||
ldap_filter = f"(sAMAccountName={target_user})"
|
||||
ldap_filter = "(sAMAccountName={})".format(target_user)
|
||||
|
||||
# Подключаемся к AD
|
||||
entries = search_by_filter(ldap_filter, ["cn"])
|
||||
|
||||
if not entries:
|
||||
return {"success": False, "error": "Пользователь {} не найден в AD.".format(target_user)}
|
||||
|
||||
user_dn = entries[0].entry_dn
|
||||
|
||||
# Подключаемся для модификации
|
||||
server = Server(AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True)
|
||||
|
||||
# Находим пользователя в дереве (нам нужен его точный путь - DN)
|
||||
conn.search(search_base=AD_BASE, search_filter=ldap_filter, attributes=["cn"])
|
||||
|
||||
if not conn.entries:
|
||||
return {"success": False, "error": f"Пользователь {target_user} не найден в AD."}
|
||||
|
||||
user_dn = conn.entries[0].entry_dn
|
||||
|
||||
# 🪄 МАГИЯ: Перезаписываем атрибут thumbnailPhoto нашими байтами
|
||||
conn.modify(user_dn, {'thumbnailPhoto': [(MODIFY_REPLACE, [photo_bytes])]})
|
||||
|
||||
|
||||
@@ -7,7 +7,8 @@ from PIL import Image
|
||||
|
||||
# Подтягиваем ваши настройки из конфига
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE
|
||||
from config.config import AD_SERVER, AD_USER, AD_PASSWORD
|
||||
from utils.ad_search import search_by_filter as ad_search_all_bases
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
|
||||
@@ -21,22 +22,14 @@ def check_user_photo(target_user: str):
|
||||
ldap_filter = f"(sAMAccountName={target_user})"
|
||||
|
||||
try:
|
||||
# Подключаемся к AD
|
||||
server = Server(AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True)
|
||||
# Подключаемся к AD и ищем по всем OU из AD_BASES
|
||||
entries = ad_search_all_bases(ldap_filter, ["cn", "thumbnailPhoto"])
|
||||
|
||||
# Ищем пользователя и просим вернуть атрибуты: имя (cn) и фото (thumbnailPhoto)
|
||||
conn.search(
|
||||
search_base=AD_BASE,
|
||||
search_filter=ldap_filter,
|
||||
attributes=["cn", "thumbnailPhoto"]
|
||||
)
|
||||
|
||||
if not conn.entries:
|
||||
if not entries:
|
||||
print("❌ Пользователь с такими данными не найден в AD.")
|
||||
return
|
||||
|
||||
user = conn.entries[0]
|
||||
user = entries[0]
|
||||
user_name = user.cn.value if 'cn' in user else target_user
|
||||
print(f"👤 Найден сотрудник: {user_name}")
|
||||
|
||||
|
||||
@@ -72,11 +72,10 @@ sd_sessions = {}
|
||||
# ================================================
|
||||
def get_ad_user_sync(login: str):
|
||||
try:
|
||||
server = Server(AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True)
|
||||
conn.search(search_base=AD_BASE, search_filter=f"(sAMAccountName={login})", attributes=["displayName", "mail", "l", "userAccountControl"])
|
||||
if conn.entries:
|
||||
user = conn.entries[0]
|
||||
from utils.ad_search import search_by_login
|
||||
entries = search_by_login(login, ["displayName", "mail", "l", "userAccountControl"])
|
||||
if entries:
|
||||
user = entries[0]
|
||||
uac = user.userAccountControl.value if 'userAccountControl' in user else 0
|
||||
return {
|
||||
"name": user.displayName.value if 'displayName' in user else login,
|
||||
|
||||
@@ -45,11 +45,10 @@ ALLOWED_EXTENSIONS = [
|
||||
# --- Функция получения Email из Active Directory ---
|
||||
def get_user_email_sync(login: str) -> str:
|
||||
try:
|
||||
server = Server(AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True)
|
||||
conn.search(search_base=AD_BASE, search_filter=f"(sAMAccountName={login})", attributes=["mail"])
|
||||
if conn.entries and 'mail' in conn.entries[0] and conn.entries[0].mail.value:
|
||||
return str(conn.entries[0].mail.value)
|
||||
from utils.ad_search import search_by_login
|
||||
entries = search_by_login(login, ["mail"])
|
||||
if entries and 'mail' in entries[0] and entries[0].mail.value:
|
||||
return str(entries[0].mail.value)
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка поиска email в AD: {e}")
|
||||
return DEFAULT_REQUESTER
|
||||
|
||||
+6
-20
@@ -3,34 +3,20 @@
|
||||
|
||||
|
||||
from ldap3 import Server, Connection, ALL
|
||||
|
||||
|
||||
def _get_ad_connection():
|
||||
"""Создать соединение с AD."""
|
||||
from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE # noqa: E501
|
||||
server = Server(AD_SERVER, get_info=ALL)
|
||||
return Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True), AD_BASE
|
||||
from utils.ad_search import search_by_filter as ad_search_all_bases
|
||||
|
||||
|
||||
def get_user_groups(cn: str) -> list[str] | None:
|
||||
"""
|
||||
Найти пользователя по CN и вернуть список групп.
|
||||
Ищет по всем OU из AD_BASES.
|
||||
Возвращает None если пользователь не найден.
|
||||
"""
|
||||
conn, ad_base = _get_ad_connection()
|
||||
try:
|
||||
conn.search(
|
||||
search_base=ad_base,
|
||||
search_filter=f"(cn={cn})",
|
||||
attributes=["cn", "sAMAccountName", "memberOf"],
|
||||
)
|
||||
if not conn.entries:
|
||||
return None
|
||||
entry = conn.entries[0]
|
||||
return [str(g) for g in entry.memberOf]
|
||||
except Exception as e:
|
||||
print(f"❌ Ошибка AD при поиске {cn}: {e}")
|
||||
entries = ad_search_all_bases("(cn={})".format(cn), ["cn", "sAMAccountName", "memberOf"])
|
||||
if not entries:
|
||||
return None
|
||||
entry = entries[0]
|
||||
return [str(g) for g in entry.memberOf]
|
||||
|
||||
|
||||
def check_group_membership(cn: str, group_pattern: str = "2FA") -> dict:
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
# /opt/trueconf_bot/utils/ad_search.py
|
||||
# Универсальный поиск пользователей по нескольким OU в AD.
|
||||
# Каждая функция принимает user_id/email/логин и ищет по всем OU из AD_BASES.
|
||||
# Возвращает первую найденную запись (ldap3 Entry) или None.
|
||||
|
||||
from ldap3 import Server, Connection, ALL
|
||||
from ldap3.utils.conv import escape_filter_chars
|
||||
import config.config as config
|
||||
|
||||
|
||||
def _get_conn():
|
||||
"""Создать соединение с AD."""
|
||||
server = Server(config.AD_SERVER, get_info=ALL)
|
||||
return Connection(server, user=config.AD_USER, password=config.AD_PASSWORD, auto_bind=True)
|
||||
|
||||
|
||||
def search_by_login(login: str, attributes: list[str] | None = None) -> list:
|
||||
"""
|
||||
Найти пользователя по sAMAccountName во всех OU из AD_BASES.
|
||||
Возвращает список Entry (обычно 0 или 1).
|
||||
"""
|
||||
attrs = attributes or ["cn", "sAMAccountName"]
|
||||
conn = _get_conn()
|
||||
for base in config.AD_BASES:
|
||||
conn.search(
|
||||
search_base=base,
|
||||
search_filter="({}={})".format("sAMAccountName", login),
|
||||
attributes=attrs,
|
||||
)
|
||||
if conn.entries:
|
||||
return list(conn.entries)
|
||||
return []
|
||||
|
||||
|
||||
def search_by_user_id(user_id: str, attributes: list[str] | None = None) -> list:
|
||||
"""
|
||||
Найти пользователя по email/UPN/sAMAccountName во всех OU из AD_BASES.
|
||||
Возвращает список Entry (обычно 0 или 1).
|
||||
"""
|
||||
attrs = attributes or ["cn", "sAMAccountName", "mail", "userPrincipalName", "extensionAttribute2"]
|
||||
safe_id = escape_filter_chars(user_id)
|
||||
short_username = user_id.split("@")[0] if "@" in user_id else user_id
|
||||
safe_sam = escape_filter_chars(short_username)
|
||||
|
||||
# Строим LDAP filter: (&(objectClass=user)(|(mail=...)(userPrincipalName=...)(sAMAccountName=...)))
|
||||
or_parts = [
|
||||
"(mail={})".format(safe_id),
|
||||
"(userPrincipalName={})".format(safe_id),
|
||||
"(userPrincipalName={})".format(short_username + "@sibcem.ru"),
|
||||
"(sAMAccountName={})".format(safe_sam),
|
||||
]
|
||||
or_filter = "({})".format("".join(or_parts))
|
||||
search_filter = "(&{}{})".format("(objectClass=user)", or_filter)
|
||||
|
||||
conn = _get_conn()
|
||||
for base in config.AD_BASES:
|
||||
conn.search(search_base=base, search_filter=search_filter, attributes=attrs)
|
||||
if conn.entries:
|
||||
return list(conn.entries)
|
||||
return []
|
||||
|
||||
|
||||
def search_by_filter(search_filter: str, attributes: list[str]) -> list:
|
||||
"""
|
||||
Найти по произвольному LDAP-фильтру во всех OU из AD_BASES.
|
||||
Возвращает список Entry.
|
||||
"""
|
||||
conn = _get_conn()
|
||||
for base in config.AD_BASES:
|
||||
conn.search(search_base=base, search_filter=search_filter, attributes=attributes)
|
||||
if conn.entries:
|
||||
return list(conn.entries)
|
||||
return []
|
||||
+9
-11
@@ -63,18 +63,16 @@ async def send_otp_via_api(phone: str, code: str) -> tuple:
|
||||
return False, str(e)
|
||||
|
||||
def get_card_id_from_ad(user_id: str) -> str:
|
||||
"""Синхронно вытягивает CARD_ID сотрудника из Active Directory (extensionAttribute2)"""
|
||||
"""Синхронно вытягивает CARD_ID сотрудника из Active Directory (extensionAttribute2).
|
||||
Ищет по всем OU из AD_BASES."""
|
||||
from utils.ad_search import search_by_user_id
|
||||
try:
|
||||
server = Server(config.AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=config.AD_USER, password=config.AD_PASSWORD, auto_bind=True)
|
||||
safe_user_id = escape_filter_chars(user_id)
|
||||
short_username = user_id.split('@')[0]
|
||||
|
||||
search_filter = f"(&(objectClass=user)(|(mail={safe_user_id})(userPrincipalName={safe_user_id})(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')})(sAMAccountName={escape_filter_chars(short_username)})))"
|
||||
conn.search(config.AD_BASE, search_filter, attributes=['extensionAttribute2'])
|
||||
|
||||
if conn.entries and 'extensionAttribute2' in conn.entries[0] and conn.entries[0].extensionAttribute2.value:
|
||||
return str(conn.entries[0].extensionAttribute2.value).strip()
|
||||
entries = search_by_user_id(user_id, ["extensionAttribute2"])
|
||||
if not entries:
|
||||
raise Exception("Пользователь не найден в AD.")
|
||||
entry = entries[0]
|
||||
if 'extensionAttribute2' in entry and entry.extensionAttribute2.value:
|
||||
return str(entry.extensionAttribute2.value).strip()
|
||||
raise Exception("Поле extensionAttribute2 не заполнено в AD.")
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка получения CARD_ID из AD для {user_id}: {e}")
|
||||
|
||||
Reference in New Issue
Block a user