""" Telegram bot for Amnezia Web Panel. Uses raw Telegram Bot API via httpx — no library version conflicts. Runs as a background asyncio task alongside the FastAPI app. """ import asyncio import html import logging import os import shlex import sys import time import uuid from typing import Optional, Callable import httpx from connection_service import ( SELF_SERVICE_PROTOCOLS, protocol_base, protocol_instance, sanitize_allowed_protocols, self_service_protocol_display_name, server_self_service_protocols, ) logger = logging.getLogger(__name__) # ----------------------------------------------------------------------- # # Global state # ----------------------------------------------------------------------- # _bot_task: Optional[asyncio.Task] = None _callback_refs = {} _pending_inputs = {} _EXTRA_PROTOCOL_DISPLAY_NAMES = { "dns": "AmneziaDNS", "socks5": "SOCKS5", "exit": "Exit Node", "adguard": "AdGuard Home", "nginx": "NGINX", } SERVICE_PROTOCOLS = {"dns", "adguard", "socks5", "nginx", "exit"} TG_TRANSLATIONS = { "en": { "hi": "Hi", "account_not_linked": "Your Telegram account is not linked to any panel user.", "contact_admin_to_link": "Please contact your administrator — they need to add your Telegram ID to your profile.", "your_telegram_id": "Your Telegram ID", "registered_admin": "You are registered as {username} with Admin role.", "choose_action": "Choose an action:", "registered_as": "You are registered as {username}.", "no_connections_create": "You have no connections yet. Tap the button below to create one!", "no_connections_contact_admin": "You have no connections yet. Please contact your administrator.", "your_connections": "Your connections ({count}) — tap to get config:", "updated": "Updated!", "access_denied": "Access denied.", "no_connections_create_short": "You have no connections. Tap the button below to create one!", "no_connections": "You have no connections.", "fetching_config": "Fetching config...", "connection_not_found": "Connection not found.", "server_not_found": "Server not found.", "fetching_config_for": "Fetching config for {name}...", "failed_retrieve_config": "Failed to retrieve configuration.", "admin_menu": "Admin menu", "action_expired": "Action expired. Use /start again.", "btn_create_connection": "Create connection", "btn_refresh_list": "Refresh list", "btn_back": "Back", "btn_do_not_assign": "Do not assign", "btn_cancel": "Cancel", "btn_servers": "Servers", "btn_users": "Users", "btn_my_connections": "My connections", "btn_how_add_server": "How to add a server", "btn_admin_menu": "Admin menu", "btn_no_installed_protocols": "No installed protocols", "btn_connections": "Connections", "btn_stop": "Stop", "btn_start": "Start", "btn_protocols": "Protocols", "btn_config": "Config", "btn_disable": "Disable", "btn_enable": "Enable", "btn_delete": "Delete", "btn_protocol": "Protocol", "choose_server": "Choose a server:", "choose_protocol": "Choose a protocol:", "protocol_label": "Protocol", "send_device_name": "Send the device name in the next message.", "cancel_instruction": "Send /cancel to cancel.", "example_label": "Example", "delete_connection_confirm": "Delete connection {name}?", "cannot_be_undone": "This cannot be undone.", "connection_deleted": "Connection deleted.", "connection_deleted_no_connections": "Connection deleted. You have no connections.", "connection_created_no_config": "Connection created, but no config is available.", "creating_connection_named": "Creating connection {name}...", "self_service_creation_disabled": "Self-service connection creation is currently disabled.\nPlease contact your administrator.", "self_service_no_servers": "No servers are available for self-service connection creation.\nPlease contact your administrator.", "self_service_disabled_contact": "Self-service is disabled. Contact your administrator.", "self_service_unavailable": "Self-service is not available.", "server_not_self_service": "This server is not available for self-service.", "no_protocols_available": "No protocols available on this server.", "name_empty": "Name cannot be empty. Send a connection name or /cancel.", "error": "Error", "saving_not_available": "Saving is not available for this bot instance.", "servers_title": "Servers", "choose_a_server": "Choose a server:", "users_title": "Users", "choose_a_user": "Choose a user:", "user_not_found": "User not found.", "protocol_not_found": "Protocol not found.", "updating_protocol": "Updating protocol container...", "loading_connections": "Loading connections...", "no_connections_short": "No connections.", "creating_connection": "Creating connection...", "updating_connection": "Updating connection...", "removing_connection": "Removing connection...", "connection_removed": "Connection removed.", "action_cancelled": "Action cancelled.", "add_server_label": "Add server", "add_server_command": "Use command:\n/addserver host username password [ssh_port] [name]\n\nExample:\n/addserver 203.0.113.10 root myPassword 22 Prod VPS\n\n⚠️ Telegram messages are not a secrets manager. Prefer adding servers in the web panel if possible.", "server_added": "Server added: {name}\nHost: {host}", "server_config": "Config", "send_connection_name": "Send the connection name in the next message.\nExample: Ivan iPhone\n\nSend /cancel to cancel.", "connection_name_label": "Connection name", "assign_to_user": "Assign this connection to a panel user?", "connection_created": "Connection created: {name}", "assigned_to": "Assigned to: {username}", "assigned_not_linked": "Assigned: not linked", "config_label": "Configuration", "connection_link_label": "Connection link", "config_part": "Configuration (part {part}/{total}):", "vpn_link_label": "VPN Link", "config_file_label": "Config file", "protocol_status": "Status", "protocol_port": "Port", "client_status": "Status", "enabled_label": "enabled", "disabled_label": "disabled", "traffic_label": "Traffic", "rx_label": "RX", "tx_label": "TX", "connections_count": "Connections", "create_new_connection": "Create connection", "connection_created_success": "Connection created successfully.", "role_label": "Role", "enabled_status": "Enabled", "email_label": "Email", "tg_id_label": "Telegram ID", "description_label": "Description", "protocol_display": "Protocol", }, "ru": { "hi": "Привет", "account_not_linked": "Ваш аккаунт Telegram не привязан ни к одному пользователю панели.", "contact_admin_to_link": "Обратитесь к администратору — он должен добавить ваш Telegram ID в профиль.", "your_telegram_id": "Ваш Telegram ID", "registered_admin": "Вы зарегистрированы как {username} с ролью Admin.", "choose_action": "Выберите действие:", "registered_as": "Вы зарегистрированы как {username}.", "no_connections_create": "У вас пока нет подключений. Нажмите кнопку ниже, чтобы создать первое.", "no_connections_contact_admin": "У вас пока нет подключений. Обратитесь к администратору.", "your_connections": "Ваши подключения ({count}) — нажмите, чтобы получить конфигурацию:", "updated": "Обновлено!", "access_denied": "Доступ запрещён.", "no_connections_create_short": "У вас нет подключений. Нажмите кнопку ниже, чтобы создать первое.", "no_connections": "У вас нет подключений.", "fetching_config": "Получаю конфигурацию...", "connection_not_found": "Подключение не найдено.", "server_not_found": "Сервер не найден.", "fetching_config_for": "Получаю конфигурацию для {name}...", "failed_retrieve_config": "Не удалось получить конфигурацию.", "admin_menu": "Меню администратора", "action_expired": "Действие устарело. Используйте /start снова.", "btn_create_connection": "Создать подключение", "btn_refresh_list": "Обновить список", "btn_back": "Назад", "btn_do_not_assign": "Не назначать", "btn_cancel": "Отмена", "btn_servers": "Серверы", "btn_users": "Пользователи", "btn_my_connections": "Мои подключения", "btn_how_add_server": "Как добавить сервер", "btn_admin_menu": "Меню администратора", "btn_no_installed_protocols": "Нет установленных протоколов", "btn_connections": "Подключения", "btn_stop": "Остановить", "btn_start": "Запустить", "btn_protocols": "Протоколы", "btn_config": "Конфигурация", "btn_disable": "Отключить", "btn_enable": "Включить", "btn_delete": "Удалить", "btn_protocol": "Протокол", "choose_server": "Выберите сервер:", "choose_protocol": "Выберите протокол:", "protocol_label": "Протокол", "send_device_name": "Отправьте имя устройства следующим сообщением.", "cancel_instruction": "Отправьте /cancel для отмены.", "example_label": "Пример", "delete_connection_confirm": "Удалить подключение {name}?", "cannot_be_undone": "Это действие нельзя отменить.", "connection_deleted": "Подключение удалено.", "connection_deleted_no_connections": "Подключение удалено. У вас нет подключений.", "connection_created_no_config": "Подключение создано, но конфигурация недоступна.", "creating_connection_named": "Создаю подключение {name}...", "self_service_creation_disabled": "Создание подключений через self-service сейчас отключено.\nОбратитесь к администратору.", "self_service_no_servers": "Нет серверов, доступных для создания подключений через self-service.\nОбратитесь к администратору.", "self_service_disabled_contact": "Self-service отключён. Обратитесь к администратору.", "self_service_unavailable": "Self-service недоступен.", "server_not_self_service": "Этот сервер недоступен для self-service.", "no_protocols_available": "На этом сервере нет доступных протоколов.", "name_empty": "Имя не может быть пустым. Отправьте имя подключения или /cancel.", "error": "Ошибка", "saving_not_available": "Сохранение недоступно для этой версии бота.", "servers_title": "Серверы", "choose_a_server": "Выберите сервер:", "users_title": "Пользователи", "choose_a_user": "Выберите пользователя:", "user_not_found": "Пользователь не найден.", "protocol_not_found": "Протокол не найден.", "updating_protocol": "Обновляю контейнер протокола...", "loading_connections": "Загружаю подключения...", "no_connections_short": "Нет подключений.", "creating_connection": "Создаю подключение...", "updating_connection": "Обновляю подключение...", "removing_connection": "Удаляю подключение...", "connection_removed": "Подключение удалено.", "action_cancelled": "Действие отменено.", "add_server_label": "Добавить сервер", "add_server_command": "Используйте команду:\n/addserver хост имя_пользователя пароль [ssh_порт] [имя]\n\nПример:\n/addserver 203.0.113.10 root myPassword 22 Prod VPS\n\n⚠️ Сообщения Telegram — не менеджер секретов. Лучше добавлять серверы через веб-панель.", "server_added": "Сервер добавлен: {name}\nХост: {host}", "server_config": "Конфигурация", "send_connection_name": "Отправьте имя подключения следующим сообщением.\nПример: Ivan iPhone\n\nОтправьте /cancel для отмены.", "connection_name_label": "Имя подключения", "assign_to_user": "Назначить это подключение пользователю панели?", "connection_created": "Подключение создано: {name}", "assigned_to": "Назначено: {username}", "assigned_not_linked": "Назначено: не привязано", "config_label": "Конфигурация", "connection_link_label": "Ссылка подключения", "config_part": "Конфигурация (часть {part}/{total}):", "vpn_link_label": "VPN ссылка", "config_file_label": "Файл конфигурации", "protocol_status": "Статус", "protocol_port": "Порт", "client_status": "Статус", "enabled_label": "включён", "disabled_label": "отключён", "traffic_label": "Трафик", "rx_label": "Входящий", "tx_label": "Исходящий", "connections_count": "Подключения", "create_new_connection": "Создать подключение", "connection_created_success": "Подключение успешно создано.", "role_label": "Роль", "enabled_status": "Включён", "email_label": "Email", "tg_id_label": "Telegram ID", "description_label": "Описание", "protocol_display": "Протокол", }, } # ----------------------------------------------------------------------- # # Public lifecycle # ----------------------------------------------------------------------- # def is_running() -> bool: return _bot_task is not None and not _bot_task.done() def launch_bot(token: str, load_data_fn: Callable, generate_vpn_link_fn: Callable, save_data_fn: Optional[Callable] = None, self_service_svc=None): global _bot_task _bot_task = asyncio.create_task( _run_bot(token, load_data_fn, generate_vpn_link_fn, save_data_fn, self_service_svc), name="telegram_bot", ) return _bot_task async def stop_bot(): global _bot_task if _bot_task and not _bot_task.done(): _bot_task.cancel() try: await _bot_task except asyncio.CancelledError: pass _bot_task = None logger.info("Telegram bot stopped.") # ----------------------------------------------------------------------- # # Low-level Telegram API helpers # ----------------------------------------------------------------------- # class TelegramAPI: def __init__(self, token: str, client: httpx.AsyncClient): self.base = f"https://api.telegram.org/bot{token}" self.client = client async def call(self, method: str, **params) -> dict: r = await self.client.post(f"{self.base}/{method}", json=params, timeout=30) return r.json() async def get_updates(self, offset: int = 0, timeout: int = 25) -> list: r = await self.client.post( f"{self.base}/getUpdates", json={"offset": offset, "timeout": timeout, "allowed_updates": ["message", "callback_query"]}, timeout=timeout + 10, ) data = r.json() if data.get("ok"): return data["result"] return [] async def send_message(self, chat_id, text: str, reply_markup=None, parse_mode="HTML") -> dict: import json params = {"chat_id": chat_id, "text": text, "parse_mode": parse_mode} if reply_markup: params["reply_markup"] = json.dumps(reply_markup) return await self.call("sendMessage", **params) async def edit_message(self, chat_id, message_id, text: str, reply_markup=None, parse_mode="HTML"): import json params = {"chat_id": chat_id, "message_id": message_id, "text": text, "parse_mode": parse_mode} if reply_markup: params["reply_markup"] = json.dumps(reply_markup) await self.call("editMessageText", **params) async def answer_callback(self, callback_query_id: str, text: str = ""): await self.call("answerCallbackQuery", callback_query_id=callback_query_id, text=text) async def send_document(self, chat_id, filename: str, content: bytes, caption: str = ""): files = {"document": (filename, content, "text/plain")} data = {"chat_id": str(chat_id), "caption": caption} r = await self.client.post(f"{self.base}/sendDocument", data=data, files=files, timeout=30) return r.json() async def _set_default_commands(api: TelegramAPI): await api.call( "setMyCommands", commands=[ {"command": "start", "description": "Open bot menu"}, {"command": "connections", "description": "Show my connections"}, {"command": "connect", "description": "Create a new connection"}, {"command": "disconnect", "description": "Delete a connection"}, ], ) # ----------------------------------------------------------------------- # # Generic helpers # ----------------------------------------------------------------------- # def _e(value) -> str: return html.escape(str(value if value is not None else "")) def _tg_lang(from_user: Optional[dict]) -> str: code = str((from_user or {}).get("language_code") or "").lower() if code.startswith("ru"): return "ru" return "en" def _tt(lang: str, key: str, **kwargs) -> str: text = TG_TRANSLATIONS.get(lang, TG_TRANSLATIONS["en"]).get(key, TG_TRANSLATIONS["en"].get(key, key)) return text.format(**kwargs) if kwargs else text def _format_bytes(value) -> str: try: value = float(value or 0) except Exception: value = 0 units = ["B", "KB", "MB", "GB", "TB"] idx = 0 while value >= 1024 and idx < len(units) - 1: value /= 1024 idx += 1 if idx == 0: return f"{int(value)} {units[idx]}" return f"{value:.2f} {units[idx]}" def _protocol_display_name(protocol: str) -> str: base = protocol_base(protocol) if base in SELF_SERVICE_PROTOCOLS: return self_service_protocol_display_name(protocol) name = _EXTRA_PROTOCOL_DISPLAY_NAMES.get(base, base) idx = protocol_instance(protocol) return name if idx <= 1 else f"{name} #{idx}" def _find_user(load_data_fn: Callable, tg_id: str, username: Optional[str] = None): data = load_data_fn() tg_id_clean = str(tg_id).lstrip("@") for u in data.get("users", []): stored = str(u.get("telegramId", "") or "").lstrip("@") if stored and stored == tg_id_clean: return u return None def _pending_key(chat_id, from_id) -> str: return f"{chat_id}:{from_id}" def _is_private_chat(chat: Optional[dict]) -> bool: return (chat or {}).get("type", "private") == "private" async def _reject_non_private(api: TelegramAPI, chat_id, lang: str = "en", callback_id: Optional[str] = None): if callback_id: await api.answer_callback(callback_id, "Use a private chat") await api.send_message(chat_id, "For security, use this command in a private chat with the bot.") def _is_admin(panel_user: dict) -> bool: return str((panel_user or {}).get("role", "")).lower() == "admin" def _ref(action: str, payload: dict) -> str: """Short callback_data indirection; Telegram callback_data is limited to 64 bytes.""" key = uuid.uuid4().hex[:12] _callback_refs[key] = {"action": action, "payload": payload, "ts": time.time()} # Opportunistic cleanup. if len(_callback_refs) > 500: cutoff = time.time() - 6 * 3600 for k in [k for k, v in _callback_refs.items() if v.get("ts", 0) < cutoff]: _callback_refs.pop(k, None) return f"r:{key}" def _resolve_ref(data_str: str): if not data_str.startswith("r:"): return None return _callback_refs.get(data_str[2:]) # ----------------------------------------------------------------------- # # Self-service helpers # ----------------------------------------------------------------------- # def _self_service_settings(data: dict) -> dict: return (data.get("settings") or {}).get("self_service") or {} def _self_service_is_enabled(data: dict) -> bool: """Returns True if self-service is enabled globally in panel settings.""" return bool(_self_service_settings(data).get("enabled", False)) def _self_service_telegram_enabled(data: dict) -> bool: """Returns True if self-service is enabled and Telegram channel is active.""" ss = _self_service_settings(data) return bool(ss.get("enabled", False) and ss.get("telegram_enabled", False)) def _get_eligible_servers(data: dict, allowed_protocols=None) -> list: """Returns list of (server_id, server, available_protos) tuples for self-service.""" if allowed_protocols is None: allowed_protocols = _self_service_settings(data).get("allowed_protocols", []) allowed_bases = sanitize_allowed_protocols(allowed_protocols) servers = data.get("servers", []) eligible = [] for sid, server in enumerate(servers): if not server.get("self_service_enabled", False): continue available = server_self_service_protocols(server, allowed_bases) if available: eligible.append((sid, server, available)) return eligible def _user_create_server_keyboard(eligible, lang: str = "en") -> dict: rows = [] for sid, server, protos in eligible: name = server.get("name") or server.get("host") or f"Server {sid + 1}" proto_text = ", ".join(self_service_protocol_display_name(p) for p in protos) rows.append([{"text": f"🖥 {name} ({proto_text})", "callback_data": _ref("user_create_server", {"sid": sid})}]) rows.append([{"text": f"❌ {_tt(lang, 'btn_cancel')}", "callback_data": "user_create_cancel"}]) return {"inline_keyboard": rows} def _build_connections_keyboard(conns: list, data: dict, lang: str = "en") -> dict: """Build inline keyboard where each button = one connection. When self-service is enabled, adds delete buttons for self-service connections and a 'Create connection' button.""" rows = [] servers = data.get("servers", []) ss_enabled = _self_service_telegram_enabled(data) for c in conns: sid = c.get("server_id", 0) server_name = "Unknown" if isinstance(sid, int) and sid < len(servers): srv = servers[sid] server_name = srv.get("name") or srv.get("host", "Unknown")[:20] proto = c.get("protocol", "").upper() name = c.get("name", "Connection") label = f"🔐 {name} · {proto} · {server_name}" row = [{"text": label, "callback_data": f"cfg:{c['id']}"}] if ss_enabled and c.get("created_by") == "self_service": row.append({"text": "🗑", "callback_data": _ref("user_delete", {"conn_id": c["id"], "name": name})}) rows.append(row) if ss_enabled: rows.append([{"text": f"➕ {_tt(lang, 'btn_create_connection')}", "callback_data": "user_create"}]) rows.append([{"text": f"🔄 {_tt(lang, 'btn_refresh_list')}", "callback_data": "refresh"}]) return {"inline_keyboard": rows} def _connection_lookup(data: dict, server_id: int, proto: str) -> dict: return { c.get("client_id"): c for c in data.get("user_connections", []) if c.get("server_id") == server_id and c.get("protocol") == proto and c.get("client_id") } def _client_display_name(client: dict, conn: Optional[dict] = None) -> str: if conn and conn.get("name"): return conn.get("name") user_data = client.get("userData") or {} return ( client.get("name") or client.get("username") or user_data.get("clientName") or user_data.get("name") or str(client.get("clientId") or client.get("client_id") or client.get("id") or "Connection")[:12] ) def _user_label(user: dict) -> str: label = user.get("username") or user.get("id", "user") role = user.get("role") or "user" suffix = f" · {role}" if user.get("telegramId"): suffix += f" · tg:{user.get('telegramId')}" if user.get("enabled") is False: suffix += " · disabled" return f"{label}{suffix}" def _users_keyboard(data: dict, back_callback: str = "adm:menu", lang: str = "en") -> dict: rows = [] for user in data.get("users", [])[:40]: rows.append([{"text": f" {_user_label(user)}", "callback_data": _ref("user", {"uid": user.get("id")})}]) rows.append([{"text": f"⬅️ {_tt(lang, 'btn_back')}", "callback_data": back_callback}]) return {"inline_keyboard": rows} def _assign_user_keyboard(data: dict, server_id: int, proto: str, name: str, lang: str = "en") -> dict: rows = [[{"text": f"🚫 {_tt(lang, 'btn_do_not_assign')}", "callback_data": _ref("create_client", {"sid": server_id, "proto": proto, "name": name, "user_id": None})}]] for user in data.get("users", [])[:40]: rows.append([{"text": f" {_user_label(user)}", "callback_data": _ref("create_client", {"sid": server_id, "proto": proto, "name": name, "user_id": user.get("id")})}]) rows.append([{"text": f"❌ {_tt(lang, 'btn_cancel')}", "callback_data": _ref("clients", {"sid": server_id, "proto": proto})}]) return {"inline_keyboard": rows} def _admin_main_keyboard(lang: str = "en") -> dict: return { "inline_keyboard": [ [{"text": f"🖥 {_tt(lang, 'btn_servers')}", "callback_data": "adm:servers"}], [{"text": f"👤 {_tt(lang, 'btn_users')}", "callback_data": "adm:users"}], [{"text": f"🔐 {_tt(lang, 'btn_my_connections')}", "callback_data": "adm:myconns"}], [{"text": f"➕ {_tt(lang, 'btn_how_add_server')}", "callback_data": "adm:addserver_help"}], ] } def _server_keyboard(data: dict, lang: str = "en") -> dict: rows = [] for sid, srv in enumerate(data.get("servers", [])): name = srv.get("name") or srv.get("host") or f"Server {sid + 1}" rows.append([{"text": f"🖥 {name}", "callback_data": f"srv:{sid}"}]) rows.append([{"text": f"⬅️ {_tt(lang, 'btn_admin_menu')}", "callback_data": "adm:menu"}]) return {"inline_keyboard": rows} def _protocol_status_icon(info: dict) -> str: if info.get("status_error"): return "⚪" running = info.get("container_running") if running is True: return "🟢" if running is False: return "🔴" return "⚪" def _protocol_status_text(info: dict, lang: str = "en") -> str: if info.get("status_error"): return "unknown ⚪" running = info.get("container_running") if running is True: return "running 🟢" if running is False: return "stopped 🔴" return "unknown ⚪" def _protocols_keyboard(server_id: int, server: dict, lang: str = "en") -> dict: rows = [] protocols = server.get("protocols", {}) or {} for proto, info in protocols.items(): installed = "✅" if info.get("installed", True) else "" rows.append([{"text": f"{installed}{_protocol_status_icon(info)} {_protocol_display_name(proto)}", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]) if not rows: rows.append([{"text": _tt(lang, "btn_no_installed_protocols"), "callback_data": f"noop"}]) rows.append([{"text": f"️ {_tt(lang, 'btn_servers')}", "callback_data": "adm:servers"}]) return {"inline_keyboard": rows} def _protocol_keyboard(server_id: int, proto: str, proto_info: dict, lang: str = "en") -> dict: base = protocol_base(proto) rows = [] if base in SELF_SERVICE_PROTOCOLS: rows.append([{"text": f" {_tt(lang, 'btn_connections')}", "callback_data": _ref("clients", {"sid": server_id, "proto": proto})}]) rows.append([{"text": f" {_tt(lang, 'btn_create_connection')}", "callback_data": _ref("add_client", {"sid": server_id, "proto": proto})}]) is_running = proto_info.get("container_running") is True rows.append([{"text": f"Stop" if is_running else f"Start", "callback_data": _ref("toggle_proto", {"sid": server_id, "proto": proto, "start": not is_running})}]) rows.append([{"text": f" {_tt(lang, 'btn_protocols')}", "callback_data": f"srv:{server_id}"}]) return {"inline_keyboard": rows} def _client_keyboard(server_id: int, proto: str, client: dict, lang: str = "en") -> dict: client_id = client.get("clientId") or client.get("client_id") or client.get("id") or "" enabled = client.get("enabled") if enabled is None: enabled = client.get("isEnabled") enabled = bool(enabled) if enabled is not None else True return { "inline_keyboard": [ [{"text": f" {_tt(lang, 'btn_config')}", "callback_data": _ref("client_cfg", {"sid": server_id, "proto": proto, "client_id": client_id, "name": client.get("name") or client.get("username") or "Connection"})}], [{"text": f"Disable" if enabled else f"Enable", "callback_data": _ref("toggle_client", {"sid": server_id, "proto": proto, "client_id": client_id, "enable": not enabled})}], [{"text": f" {_tt(lang, 'btn_delete')}", "callback_data": _ref("remove_client", {"sid": server_id, "proto": proto, "client_id": client_id})}], [{"text": f" {_tt(lang, 'btn_connections')}", "callback_data": _ref("clients", {"sid": server_id, "proto": proto})}], ] } def _get_ssh_and_manager(server: dict, proto: str): sys.path.insert(0, os.path.dirname(__file__)) from managers.ssh_manager import SSHManager from managers.awg_manager import AWGManager from managers.xray_manager import XrayManager from managers.telemt_manager import TelemtManager from managers.wireguard_manager import WireGuardManager from managers.dns_manager import DNSManager from managers.socks5_manager import Socks5Manager from managers.adguard_manager import AdguardManager from managers.nginx_manager import NginxManager from managers.exit_manager import ExitManager ssh = SSHManager( server["host"], server.get("ssh_port", 22), server["username"], server.get("password", ""), server.get("private_key", ""), ) base = protocol_base(proto) if base == "xray": manager = XrayManager(ssh, proto) elif base == "telemt": manager = TelemtManager(ssh, proto) elif base == "wireguard": manager = WireGuardManager(ssh) elif base == "dns": manager = DNSManager(ssh) elif base == "socks5": manager = Socks5Manager(ssh, proto) elif base == "adguard": manager = AdguardManager(ssh) elif base == "nginx": manager = NginxManager(ssh, proto) elif base == "exit": manager = ExitManager(ssh, proto) else: manager = AWGManager(ssh) return ssh, manager def _manager_call(manager, method_name: str, proto: str, *args, **kwargs): method = getattr(manager, method_name) try: return method(proto, *args, **kwargs) except TypeError: return method(*args, **kwargs) def _refresh_server_protocol_statuses(server: dict) -> dict: """Refresh saved protocol metadata with live Docker status for Telegram admin views.""" protocols = server.get("protocols", {}) or {} if not protocols: return server ssh = None try: ssh, _ = _get_ssh_and_manager(server, "awg") ssh.connect() for proto, info in protocols.items(): container = info.get("container_name") if not container: try: _, manager = _get_ssh_and_manager(server, proto) container = getattr(manager, "container_name", None) or getattr(manager, "CONTAINER_NAME", None) except Exception: container = None if not container: info["status_error"] = "Container name is unknown" continue out, err, code = ssh.run_sudo_command( f"docker inspect -f '{{{{.State.Running}}}}' {shlex.quote(str(container))} 2>/dev/null" ) if code == 0: info["container_running"] = out.strip().lower() == "true" info["container_exists"] = True info.pop("status_error", None) else: info["container_running"] = False info["container_exists"] = False info["status_error"] = (err or out or "container not found").strip() except Exception as e: logger.warning("Telegram bot: failed to refresh protocol statuses for %s: %s", server.get("name") or server.get("host"), e) for info in protocols.values(): info["status_error"] = str(e) finally: if ssh: try: ssh.disconnect() except Exception: pass return server async def _refresh_server_protocol_statuses_async(server: dict) -> dict: return await asyncio.to_thread(_refresh_server_protocol_statuses, server) # ----------------------------------------------------------------------- # # /start and user connection handlers # ----------------------------------------------------------------------- # async def _handle_start(api: TelegramAPI, msg: dict, load_data_fn: Callable): chat_id = msg["chat"]["id"] tg_id = str(msg["from"]["id"]) tg_username = msg["from"].get("username") first_name = msg["from"].get("first_name", "") lang = _tg_lang(msg["from"]) panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.send_message( chat_id, f"👋 {_tt(lang, 'hi')}, {_e(first_name)}!\n\n" f"{_tt(lang, 'account_not_linked')}\n" f"{_tt(lang, 'contact_admin_to_link')}\n\n" f"{_tt(lang, 'your_telegram_id')}: {_e(tg_id)}", ) return if _is_admin(panel_user): await api.send_message( chat_id, f"👋 {_tt(lang, 'hi')}, {_e(first_name)}!\n\n" f"{_tt(lang, 'registered_admin', username=_e(panel_user.get('username')))}\n" f"{_tt(lang, 'choose_action')}", reply_markup=_admin_main_keyboard(lang), ) return await _send_user_connections(api, chat_id, panel_user, load_data_fn, first_name=first_name, lang=lang) async def _send_user_connections(api: TelegramAPI, chat_id: int, panel_user: dict, load_data_fn: Callable, first_name: str = "", lang: str = "en"): data = load_data_fn() conns = [c for c in data.get("user_connections", []) if c.get("user_id") == panel_user.get("id")] if not conns: greeting = f"👋 {_tt(lang, 'hi')}, {_e(first_name)}!\n\n" if first_name else "" if _self_service_telegram_enabled(data): kb = _build_connections_keyboard(conns, data, lang) await api.send_message( chat_id, greeting + f"{_tt(lang, 'registered_as', username=_e(panel_user.get('username')))}.\n\n" f"{_tt(lang, 'no_connections_create')}", reply_markup=kb, ) else: await api.send_message( chat_id, greeting + f"{_tt(lang, 'registered_as', username=_e(panel_user.get('username')))}.\n\n" f"{_tt(lang, 'no_connections_contact_admin')}", ) return kb = _build_connections_keyboard(conns, data, lang) greeting = f"👋 {_tt(lang, 'hi')}, {_e(first_name)}!\n\n" if first_name else "" await api.send_message( chat_id, greeting + f"{_tt(lang, 'registered_as', username=_e(panel_user.get('username')))}.\n\n" f"{_tt(lang, 'your_connections', count=len(conns))}", reply_markup=kb, ) async def _handle_refresh(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, tg_id: str, tg_username: Optional[str], load_data_fn: Callable, lang: str = "en"): await api.answer_callback(callback_id, _tt(lang, "updated")) panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'access_denied')}") return data = load_data_fn() conns = [c for c in data.get("user_connections", []) if c.get("user_id") == panel_user.get("id")] if not conns: if _self_service_telegram_enabled(data): kb = _build_connections_keyboard(conns, data, lang) await api.edit_message(chat_id, message_id, _tt(lang, "no_connections_create_short"), reply_markup=kb) else: await api.edit_message(chat_id, message_id, _tt(lang, "no_connections")) return kb = _build_connections_keyboard(conns, data, lang) await api.edit_message(chat_id, message_id, _tt(lang, "your_connections", count=len(conns)), reply_markup=kb) async def _handle_get_config(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, conn_id: str, tg_id: str, tg_username: Optional[str], load_data_fn: Callable, generate_vpn_link_fn: Callable, lang: str = "en"): await api.answer_callback(callback_id, _tt(lang, "fetching_config")) panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.send_message(chat_id, f"❌ {_tt(lang, 'access_denied')}") return data = load_data_fn() conn = next((c for c in data.get("user_connections", []) if c.get("id") == conn_id and (_is_admin(panel_user) or c.get("user_id") == panel_user.get("id"))), None) if not conn: await api.send_message(chat_id, f"❌ {_tt(lang, 'connection_not_found')}") return servers = data.get("servers", []) sid = conn.get("server_id") if not isinstance(sid, int) or sid >= len(servers): await api.send_message(chat_id, f"❌ {_tt(lang, 'server_not_found')}") return await _send_config_by_client(api, chat_id, servers[sid], conn.get("protocol", "awg"), conn.get("client_id"), conn.get("name", "Connection"), generate_vpn_link_fn, lang) async def _send_config_by_client(api: TelegramAPI, chat_id: int, server: dict, proto: str, client_id: str, conn_name: str, generate_vpn_link_fn: Callable, lang: str = "en"): loading_result = await api.send_message(chat_id, _tt(lang, "fetching_config_for", name=_e(conn_name))) loading_msg_id = loading_result.get("result", {}).get("message_id") try: proto_info = server.get("protocols", {}).get(proto, {}) port = proto_info.get("port", "55424") def _get_cfg(): ssh, manager = _get_ssh_and_manager(server, proto) try: ssh.connect() return _manager_call(manager, "get_client_config", proto, client_id, server["host"], port) finally: ssh.disconnect() config = await asyncio.to_thread(_get_cfg) if not config: if loading_msg_id: await api.edit_message(chat_id, loading_msg_id, f"❌ {_tt(lang, 'failed_retrieve_config')}") return if loading_msg_id: await api.call("deleteMessage", chat_id=chat_id, message_id=loading_msg_id) server_name = server.get("name") or server.get("host", "Unknown") await api.send_message(chat_id, f"✅ {_e(conn_name)}\n🌐 {_tt(lang, 'servers_title')}: {_e(server_name)}\n🔌 {_tt(lang, 'protocol_label')}: {_e(proto.upper())}") is_link_proto = protocol_base(proto) in ("xray", "telemt") if is_link_proto: await api.send_message(chat_id, f"🔗 {_tt(lang, 'connection_link_label')} (tap to copy):\n{_e(config)}") else: MAX_LEN = 4000 if len(config) <= MAX_LEN: await api.send_message(chat_id, f"📄 {_tt(lang, 'config_label')}:\n
{_e(config)}
") else: chunks = [config[i:i + MAX_LEN] for i in range(0, len(config), MAX_LEN)] for i, chunk in enumerate(chunks, 1): await api.send_message(chat_id, f"📄 {_tt(lang, 'config_part', part=i, total=len(chunks))}:\n
{_e(chunk)}
") vpn_link = generate_vpn_link_fn(config, server, proto) if config else "" if vpn_link: await api.send_message(chat_id, f"🔗 {_tt(lang, 'vpn_link_label')} (tap to copy):\n{_e(vpn_link)}") filename = f"{str(conn_name).replace(' ', '_')}.conf" await api.send_document(chat_id, filename=filename, content=config.encode("utf-8"), caption=f"📁 {_tt(lang, 'config_file_label')}: {conn_name}") except Exception as e: logger.exception("Bot: error getting config") if loading_msg_id: await api.edit_message(chat_id, loading_msg_id, f"❌ {_tt(lang, 'error')}: {_e(e)}") else: await api.send_message(chat_id, f"❌ {_tt(lang, 'error')}: {_e(e)}") # ----------------------------------------------------------------------- # # Admin handlers # ----------------------------------------------------------------------- # def _require_admin(load_data_fn: Callable, tg_id: str, tg_username: Optional[str] = None): user = _find_user(load_data_fn, tg_id, tg_username) if not user or not _is_admin(user): return None return user async def _handle_add_server_command(api: TelegramAPI, msg: dict, load_data_fn: Callable, save_data_fn: Optional[Callable], lang: str = "en"): chat_id = msg["chat"]["id"] tg_id = str(msg["from"]["id"]) if not _require_admin(load_data_fn, tg_id): await api.send_message(chat_id, f"❌ {_tt(lang, 'access_denied')}") return if not save_data_fn: await api.send_message(chat_id, f"❌ {_tt(lang, 'saving_not_available')}") return text = msg.get("text", "") parts = text.split(maxsplit=5) if len(parts) < 4: await api.send_message( chat_id, _tt(lang, "add_server_command"), ) return host = parts[1] username = parts[2] password = parts[3] ssh_port = 22 name = host if len(parts) >= 5: try: ssh_port = int(parts[4]) except Exception: name = parts[4] if len(parts) >= 6: name = parts[5] or host data = load_data_fn() data.setdefault("servers", []).append({ "name": name, "host": host, "ssh_port": ssh_port, "username": username, "password": password, "private_key": "", "protocols": {}, "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"), }) save_data_fn(data) await api.send_message(chat_id, _tt(lang, "server_added", name=_e(name), host=_e(host))) async def _admin_servers(api: TelegramAPI, chat_id: int, message_id: Optional[int], load_data_fn: Callable, lang: str = "en"): data = load_data_fn() servers = data.get("servers", []) text = f"🖥 {_tt(lang, 'servers_title')} ({len(servers)})\n\n{_tt(lang, 'choose_a_server')}" if message_id: await api.edit_message(chat_id, message_id, text, reply_markup=_server_keyboard(data, lang)) else: await api.send_message(chat_id, text, reply_markup=_server_keyboard(data, lang)) async def _admin_users(api: TelegramAPI, chat_id: int, message_id: int, load_data_fn: Callable, lang: str = "en"): data = load_data_fn() users = data.get("users", []) await api.edit_message(chat_id, message_id, f"👤 {_tt(lang, 'users_title')} ({len(users)})\n\n{_tt(lang, 'choose_a_user')}", reply_markup=_users_keyboard(data, lang)) async def _admin_user_detail(api: TelegramAPI, chat_id: int, message_id: int, user_id: str, load_data_fn: Callable, lang: str = "en"): data = load_data_fn() user = next((u for u in data.get("users", []) if u.get("id") == user_id), None) if not user: await api.edit_message(chat_id, message_id, "❌ User not found.", reply_markup={"inline_keyboard": [[{"text": "⬅️ Users", "callback_data": "adm:users"}]]}) return conns = [c for c in data.get("user_connections", []) if c.get("user_id") == user_id] lines = [ f"👤 {_e(user.get('username'))}", f"{_tt(lang, 'role_label')}: {_e(user.get('role', 'user'))}", f"{_tt(lang, 'enabled_status')}: {'yes ✅' if user.get('enabled', True) else 'no 🚫'}", f"{_tt(lang, 'tg_id_label')}: {_e(user.get('telegramId') or '-')}", f"{_tt(lang, 'email_label')}: {_e(user.get('email') or '-')}", f"{_tt(lang, 'connections_count')}: {len(conns)}", ] if user.get("description"): lines.append(f"{_tt(lang, 'description_label')}: {_e(user.get('description'))}") rows = [] servers = data.get("servers", []) for c in conns[:20]: sid = c.get("server_id") server_name = "Unknown" if isinstance(sid, int) and sid < len(servers): server_name = servers[sid].get("name") or servers[sid].get("host") or "Unknown" rows.append([{"text": f"🔐 {c.get('name', 'Connection')} · {c.get('protocol', '').upper()} · {server_name}", "callback_data": f"cfg:{c.get('id')}"}]) rows.append([{"text": f"⬅️ {_tt(lang, 'btn_users')}", "callback_data": "adm:users"}]) rows.append([{"text": f"⬅️ {_tt(lang, 'btn_admin_menu')}", "callback_data": "adm:menu"}]) await api.edit_message(chat_id, message_id, "\n".join(lines), reply_markup={"inline_keyboard": rows}) async def _admin_server_detail(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, load_data_fn: Callable, lang: str = "en"): data = load_data_fn() servers = data.get("servers", []) if server_id < 0 or server_id >= len(servers): await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'server_not_found')}") return server = await _refresh_server_protocol_statuses_async(servers[server_id]) protocols = server.get("protocols", {}) or {} text = ( f"🖥 {_e(server.get('name') or server.get('host'))}\n" f"Host: {_e(server.get('host'))}\n" f"SSH: {_e(server.get('username'))}@{_e(server.get('host'))}:{_e(server.get('ssh_port', 22))}\n\n" f"{_tt(lang, 'btn_protocols')} ({len(protocols)}):" ) await api.edit_message(chat_id, message_id, text, reply_markup=_protocols_keyboard(server_id, server, lang)) async def _admin_protocol_detail(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, load_data_fn: Callable, lang: str = "en"): data = load_data_fn() servers = data.get("servers", []) if server_id < 0 or server_id >= len(servers): await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'server_not_found')}") return server = await _refresh_server_protocol_statuses_async(servers[server_id]) info = (server.get("protocols", {}) or {}).get(proto) if not info: await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'protocol_not_found')}") return lines = [ f" {_e(_protocol_display_name(proto))}", f"{_tt(lang, 'servers_title')}: {_e(server.get('name') or server.get('host'))}", f"{_tt(lang, 'protocol_status')}: {_protocol_status_text(info, lang)}", ] exit_link = info.get("exit_link") or {} if exit_link: lines.append(f"exit: {_e(exit_link.get('exit_name'))}" + (f" ({_e(exit_link.get('stale'))})" if exit_link.get("stale") else "")) for key in ("port", "container_name", "domain", "site_url", "web_port", "mode", "subnet", "peers_count"): if info.get(key) not in (None, ""): lines.append(f"{_e(key)}: {_e(info.get(key))}") if info.get("status_error"): lines.append(f"status_error: {_e(info.get('status_error'))}") await api.edit_message(chat_id, message_id, "\n".join(lines), reply_markup=_protocol_keyboard(server_id, proto, info, lang)) async def _admin_toggle_protocol(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, start: bool, load_data_fn: Callable, lang: str = "en"): await api.edit_message(chat_id, message_id, f"⏳ {_tt(lang, 'updating_protocol')}") def _toggle(): data = load_data_fn() server = data["servers"][server_id] ssh, manager = _get_ssh_and_manager(server, proto) try: ssh.connect() container = (server.get("protocols", {}).get(proto, {}) or {}).get("container_name") if not container: # fallback: most managers expose CONTAINER_NAME for base/first instances container = getattr(manager, "CONTAINER_NAME", None) if not container: raise RuntimeError("Container name is unknown") action = "start" if start else "stop" out, err, code = ssh.run_sudo_command(f"docker {action} {container}") if code != 0: raise RuntimeError(err or out or f"docker {action} failed") return data finally: ssh.disconnect() try: await asyncio.to_thread(_toggle) await _admin_protocol_detail(api, chat_id, message_id, server_id, proto, load_data_fn, lang) except Exception as e: logger.exception("Bot admin: protocol toggle failed") await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'error')}: {_e(e)}", reply_markup={"inline_keyboard": [[{"text": f"⬅️ {_tt(lang, 'btn_protocol')}", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]]}) async def _admin_clients(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, load_data_fn: Callable, lang: str = "en"): await api.edit_message(chat_id, message_id, f"⏳ {_tt(lang, 'loading_connections')}") def _load_clients(): data = load_data_fn() server = data["servers"][server_id] ssh, manager = _get_ssh_and_manager(server, proto) try: ssh.connect() return data, _manager_call(manager, "get_clients", proto) finally: ssh.disconnect() try: data, clients = await asyncio.to_thread(_load_clients) if not clients: await api.edit_message(chat_id, message_id, f"👥 {_tt(lang, 'no_connections_short')}", reply_markup={"inline_keyboard": [[{"text": f"➕ {_tt(lang, 'btn_create_connection')}", "callback_data": _ref("add_client", {"sid": server_id, "proto": proto})}], [{"text": f"⬅️ {_tt(lang, 'btn_protocol')}", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]]}) return rows = [] conn_by_client = _connection_lookup(data, server_id, proto) users_by_id = {u.get("id"): u for u in data.get("users", [])} for c in clients[:40]: client_id = c.get("clientId") or c.get("client_id") or c.get("id") or "" conn = conn_by_client.get(client_id) name = _client_display_name(c, conn) traffic = "" user_data = c.get("userData") or {} if user_data: total = (user_data.get("dataReceivedBytes") or 0) + (user_data.get("dataSentBytes") or 0) traffic = f" · {_format_bytes(total)}" assigned = "" if conn and conn.get("user_id") in users_by_id: assigned = f" · @{users_by_id[conn.get('user_id')].get('username')}" c["name"] = name c["assigned_user_id"] = conn.get("user_id") if conn else None rows.append([{"text": f" {name}{assigned}{traffic}", "callback_data": _ref("client", {"sid": server_id, "proto": proto, "client_id": client_id, "name": name, "client": c})}]) rows.append([{"text": f"➕ {_tt(lang, 'btn_create_connection')}", "callback_data": _ref("add_client", {"sid": server_id, "proto": proto})}]) rows.append([{"text": f"⬅️ {_tt(lang, 'btn_protocol')}", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]) await api.edit_message(chat_id, message_id, f"👥 {_e(_protocol_display_name(proto))} {_tt(lang, 'connections_count')} ({len(clients)})", reply_markup={"inline_keyboard": rows}) except Exception as e: logger.exception("Bot admin: load clients failed") await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'error')}: {_e(e)}", reply_markup={"inline_keyboard": [[{"text": f"️ {_tt(lang, 'btn_protocol')}", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]]}) async def _admin_client_detail(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, client: dict, lang: str = "en"): client_id = client.get("clientId") or client.get("client_id") or client.get("id") or "" name = _client_display_name(client) user_data = client.get("userData") or {} rx = user_data.get("dataReceivedBytes") or 0 tx = user_data.get("dataSentBytes") or 0 enabled = client.get("enabled") if enabled is None: enabled = client.get("isEnabled") enabled_text = f"{_tt(lang, 'enabled_label')} ✅" if (enabled is None or enabled) else f"{_tt(lang, 'disabled_label')} 🚫" text = ( f"👤 {_e(name)}\n" f"{_tt(lang, 'protocol_label')}: {_e(_protocol_display_name(proto))}\n" f"Client ID: {_e(client_id)}\n" f"{_tt(lang, 'client_status')}: {enabled_text}" ) if user_data: text += f"\n{_tt(lang, 'traffic_label')}: {_format_bytes(rx + tx)}\n{_tt(lang, 'rx_label')}: {_format_bytes(rx)} · {_tt(lang, 'tx_label')}: {_format_bytes(tx)}" await api.edit_message(chat_id, message_id, text, reply_markup=_client_keyboard(server_id, proto, client, lang)) async def _admin_add_client(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, panel_user: dict, load_data_fn: Callable, save_data_fn: Optional[Callable], generate_vpn_link_fn: Callable, lang: str = "en", tg_id: Optional[str] = None): if not save_data_fn: await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'saving_not_available')}") return _pending_inputs[_pending_key(chat_id, tg_id or panel_user.get('telegramId') or panel_user.get('id'))] = { "kind": "add_client_name", "sid": server_id, "proto": proto, "admin_user_id": panel_user.get("id"), "ts": time.time(), } await api.edit_message( chat_id, message_id, f"➕ {_tt(lang, 'create_new_connection')}\n\n" f"{_tt(lang, 'servers_title')}/{_tt(lang, 'protocol_label')}: {_e(_protocol_display_name(proto))}\n\n" f"{_tt(lang, 'send_connection_name')}", reply_markup={"inline_keyboard": [[{"text": f"❌ {_tt(lang, 'btn_cancel')}", "callback_data": _ref("clients", {"sid": server_id, "proto": proto})}]]}, ) async def _admin_choose_client_user(api: TelegramAPI, chat_id: int, name: str, server_id: int, proto: str, load_data_fn: Callable, lang: str = "en"): data = load_data_fn() await api.send_message( chat_id, f"✅ {_tt(lang, 'connection_name_label')}: {_e(name)}\n\n" f"{_tt(lang, 'assign_to_user')}", reply_markup=_assign_user_keyboard(data, server_id, proto, name, lang), ) async def _admin_create_client(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, name: str, user_id: Optional[str], load_data_fn: Callable, save_data_fn: Optional[Callable], generate_vpn_link_fn: Callable, lang: str = "en"): if not save_data_fn: await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'saving_not_available')}") return await api.edit_message(chat_id, message_id, f"⏳ {_tt(lang, 'creating_connection')}") def _create(): data = load_data_fn() server = data["servers"][server_id] proto_info = (server.get("protocols", {}) or {}).get(proto, {}) port = proto_info.get("port", "55424") ssh, manager = _get_ssh_and_manager(server, proto) try: ssh.connect() if protocol_base(proto) == "telemt": result = manager.add_client(proto, name, server["host"], port) elif protocol_base(proto) == "wireguard": result = manager.add_client(name, server["host"]) else: result = manager.add_client(proto, name, server["host"], port) finally: ssh.disconnect() client_id = result.get("client_id") or result.get("clientId") assigned_user = None if user_id: assigned_user = next((u for u in data.get("users", []) if u.get("id") == user_id), None) if user_id and client_id: data.setdefault("user_connections", []).append({ "id": str(uuid.uuid4()), "user_id": user_id, "server_id": server_id, "protocol": proto, "client_id": client_id, "name": name, "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"), }) save_data_fn(data) return server, result, client_id, assigned_user try: server, result, client_id, assigned_user = await asyncio.to_thread(_create) assigned_text = f"\n{_tt(lang, 'assigned_to', username=_e(assigned_user.get('username')))}" if assigned_user else f"\n{_tt(lang, 'assigned_not_linked')}" await api.edit_message(chat_id, message_id, _tt(lang, "connection_created", name=_e(name)) + assigned_text) config = result.get("config") if config: await _send_config_text(api, chat_id, server, proto, name, config, generate_vpn_link_fn, lang) elif client_id: await _send_config_by_client(api, chat_id, server, proto, client_id, name, generate_vpn_link_fn, lang) except Exception as e: logger.exception("Bot admin: add client failed") await api.edit_message(chat_id, message_id, f" {_tt(lang, 'error')}: {_e(e)}", reply_markup={"inline_keyboard": [[{"text": f"⬅️ {_tt(lang, 'btn_protocol')}", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]]}) async def _send_config_text(api: TelegramAPI, chat_id: int, server: dict, proto: str, conn_name: str, config: str, generate_vpn_link_fn: Callable, lang: str = "en"): await api.send_message(chat_id, f"✅ {_e(conn_name)}\n🌐 {_tt(lang, 'servers_title')}: {_e(server.get('name') or server.get('host'))}\n {_tt(lang, 'protocol_label')}: {_e(proto.upper())}") if protocol_base(proto) in ("xray", "telemt"): await api.send_message(chat_id, f"🔗 {_tt(lang, 'connection_link_label')}:\n{_e(config)}") else: await api.send_message(chat_id, f"📄 Configuration:\n
{_e(config)}
") vpn_link = generate_vpn_link_fn(config, server, proto) if config else "" if vpn_link: await api.send_message(chat_id, f" {_tt(lang, 'vpn_link_label')}:\n{_e(vpn_link)}") await api.send_document(chat_id, filename=f"{conn_name}.conf", content=config.encode("utf-8"), caption=f" {_tt(lang, 'config_file_label')}: {conn_name}") async def _admin_toggle_client(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, client_id: str, enable: bool, load_data_fn: Callable, lang: str = "en"): await api.edit_message(chat_id, message_id, f" {_tt(lang, 'updating_connection')}") def _toggle(): data = load_data_fn() server = data["servers"][server_id] ssh, manager = _get_ssh_and_manager(server, proto) try: ssh.connect() return _manager_call(manager, "toggle_client", proto, client_id, enable) finally: ssh.disconnect() try: await asyncio.to_thread(_toggle) await api.edit_message(chat_id, message_id, f"✅ {_tt(lang, 'updated')}", reply_markup={"inline_keyboard": [[{"text": f"⬅️ {_tt(lang, 'btn_connections')}", "callback_data": _ref("clients", {"sid": server_id, "proto": proto})}]]}) except Exception as e: logger.exception("Bot admin: toggle client failed") await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'error')}: {_e(e)}") async def _admin_remove_client(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, client_id: str, load_data_fn: Callable, save_data_fn: Optional[Callable], lang: str = "en"): if not save_data_fn: await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'saving_not_available')}") return await api.edit_message(chat_id, message_id, f" {_tt(lang, 'removing_connection')}") def _remove(): data = load_data_fn() server = data["servers"][server_id] ssh, manager = _get_ssh_and_manager(server, proto) try: ssh.connect() _manager_call(manager, "remove_client", proto, client_id) finally: ssh.disconnect() data["user_connections"] = [ c for c in data.get("user_connections", []) if not (c.get("server_id") == server_id and c.get("protocol") == proto and c.get("client_id") == client_id) ] save_data_fn(data) try: await asyncio.to_thread(_remove) await api.edit_message(chat_id, message_id, f"✅ {_tt(lang, 'connection_removed')}", reply_markup={"inline_keyboard": [[{"text": f"⬅️ {_tt(lang, 'btn_connections')}", "callback_data": _ref("clients", {"sid": server_id, "proto": proto})}]]}) except Exception as e: logger.exception("Bot admin: remove client failed") await api.edit_message(chat_id, message_id, f" {_tt(lang, 'error')}: {_e(e)}") async def _handle_pending_input(api: TelegramAPI, msg: dict, load_data_fn: Callable, save_data_fn: Optional[Callable], generate_vpn_link_fn: Callable, self_service_svc=None) -> bool: chat_id = msg["chat"]["id"] from_id = msg["from"]["id"] lang = _tg_lang(msg.get("from")) key = _pending_key(chat_id, from_id) state = _pending_inputs.get(key) if not state: return False text = (msg.get("text") or "").strip() if text.lower() in ("/cancel", "cancel"): _pending_inputs.pop(key, None) await api.send_message(chat_id, f"❌ {_tt(lang, 'action_cancelled')}", reply_markup=_admin_main_keyboard(lang)) return True if text.startswith("/"): _pending_inputs.pop(key, None) return False if state.get("kind") == "add_client_name": panel_user = _require_admin(load_data_fn, str(msg["from"]["id"]), msg["from"].get("username")) if not panel_user: _pending_inputs.pop(key, None) await api.send_message(chat_id, f"❌ {_tt(lang, 'access_denied')}") return True name = text[:80].strip() if not name: await api.send_message(chat_id, _tt(lang, "name_empty")) return True _pending_inputs.pop(key, None) await _admin_choose_client_user(api, chat_id, name, int(state.get("sid", 0)), state.get("proto", "awg"), load_data_fn, lang) return True if state.get("kind") == "user_add_client_name": panel_user = _find_user(load_data_fn, str(msg["from"]["id"]), msg["from"].get("username")) if not panel_user: _pending_inputs.pop(key, None) await api.send_message(chat_id, f"❌ {_tt(lang, 'access_denied')}") return True name = text[:80].strip() if not name: await api.send_message(chat_id, _tt(lang, "name_empty")) return True _pending_inputs.pop(key, None) if not self_service_svc: await api.send_message(chat_id, f"❌ {_tt(lang, 'self_service_unavailable')}") return True data = load_data_fn() if not _self_service_telegram_enabled(data): await api.send_message(chat_id, _tt(lang, "self_service_creation_disabled")) return True sid = int(state.get("sid", 0)) proto = state.get("proto", "awg") await _user_create_connection(api, chat_id, panel_user, sid, proto, name, load_data_fn, generate_vpn_link_fn, self_service_svc, lang) return True return False # ----------------------------------------------------------------------- # # Self-service user create wizard # ----------------------------------------------------------------------- # async def _user_create_start(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, tg_id: str, tg_username: Optional[str], load_data_fn: Callable, self_service_svc, lang: str = "en"): panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.answer_callback(callback_id, text="Access denied") await api.edit_message(chat_id, message_id, "❌ Access denied.") return data = load_data_fn() if not _self_service_telegram_enabled(data): await api.answer_callback(callback_id) await api.edit_message( chat_id, message_id, _tt(lang, "self_service_disabled_contact"), ) return eligible = _get_eligible_servers(data) if not eligible: await api.answer_callback(callback_id) await api.edit_message( chat_id, message_id, _tt(lang, "self_service_no_servers"), ) return await api.answer_callback(callback_id) await api.edit_message( chat_id, message_id, f"➕ {_tt(lang, 'btn_create_connection')}\n\n{_tt(lang, 'choose_server')}", reply_markup=_user_create_server_keyboard(eligible, lang), ) async def _user_create_start_message(api: TelegramAPI, chat_id: int, tg_id: str, tg_username: Optional[str], load_data_fn: Callable, lang: str = "en"): panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.send_message(chat_id, f"❌ {_tt(lang, 'access_denied')}") return data = load_data_fn() if not _self_service_telegram_enabled(data): await api.send_message(chat_id, _tt(lang, "self_service_disabled_contact")) return eligible = _get_eligible_servers(data) if not eligible: await api.send_message(chat_id, _tt(lang, "self_service_no_servers")) return await api.send_message( chat_id, f"➕ {_tt(lang, 'btn_create_connection')}\n\n{_tt(lang, 'choose_server')}", reply_markup=_user_create_server_keyboard(eligible, lang), ) async def _user_create_server(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, tg_id: str, tg_username: Optional[str], server_id: int, load_data_fn: Callable, self_service_svc, lang: str = "en"): panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.answer_callback(callback_id, text="Access denied") await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'access_denied')}") return await api.answer_callback(callback_id) data = load_data_fn() if not _self_service_telegram_enabled(data): await api.edit_message(chat_id, message_id, _tt(lang, "self_service_disabled_contact")) return servers = data.get("servers", []) if server_id < 0 or server_id >= len(servers): await api.edit_message(chat_id, message_id, f" {_tt(lang, 'server_not_found')}") return server = servers[server_id] if not server.get("self_service_enabled", False): await api.edit_message(chat_id, message_id, _tt(lang, "server_not_self_service")) return allowed_bases = sanitize_allowed_protocols(_self_service_settings(data).get("allowed_protocols", [])) available_protos = server_self_service_protocols(server, allowed_bases) if not available_protos: await api.edit_message(chat_id, message_id, _tt(lang, "no_protocols_available")) return server_name = server.get("name") or server.get("host") or f"Server {server_id + 1}" rows = [] for proto in available_protos: rows.append([{"text": _protocol_display_name(proto), "callback_data": _ref("user_create_protocol", {"sid": server_id, "proto": proto})}]) rows.append([{"text": f" {_tt(lang, 'btn_back')}", "callback_data": "user_create"}]) await api.edit_message( chat_id, message_id, f" {_tt(lang, 'servers_title')}: {_e(server_name)}\n\n{_tt(lang, 'choose_protocol')}", reply_markup={"inline_keyboard": rows}, ) async def _user_create_protocol(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, tg_id: str, tg_username: Optional[str], server_id: int, proto: str, load_data_fn: Callable, lang: str = "en"): panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.answer_callback(callback_id, text="Access denied") await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'access_denied')}") return await api.answer_callback(callback_id) _pending_inputs[_pending_key(chat_id, tg_id)] = { "kind": "user_add_client_name", "sid": server_id, "proto": proto, "ts": time.time(), } await api.edit_message( chat_id, message_id, f"➕ {_tt(lang, 'btn_create_connection')}\n\n" f"{_tt(lang, 'protocol_label')}: {_e(_protocol_display_name(proto))}\n\n" f"{_tt(lang, 'send_device_name')}\n" f"{_tt(lang, 'example_label')}: Ivan iPhone\n\n" f"{_tt(lang, 'cancel_instruction')}", reply_markup={"inline_keyboard": [[{"text": f"❌ {_tt(lang, 'btn_cancel')}", "callback_data": "user_create_cancel"}]]}, ) async def _user_create_cancel(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, tg_id: str, tg_username: Optional[str], load_data_fn: Callable, lang: str = "en"): _pending_inputs.pop(_pending_key(chat_id, tg_id), None) panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.answer_callback(callback_id, text="Access denied") await api.edit_message(chat_id, message_id, f" {_tt(lang, 'access_denied')}") return await api.answer_callback(callback_id) await _send_user_connections(api, chat_id, panel_user, load_data_fn, lang=lang) async def _user_delete(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, tg_id: str, tg_username: Optional[str], conn_id: str, name: str, load_data_fn: Callable, lang: str = "en"): panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.answer_callback(callback_id, text="Access denied") await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'access_denied')}") return await api.answer_callback(callback_id) data = load_data_fn() if not _self_service_telegram_enabled(data): await api.edit_message(chat_id, message_id, _tt(lang, "self_service_disabled_contact")) return conn = next((c for c in data.get("user_connections", []) if c.get("id") == conn_id and c.get("user_id") == panel_user.get("id")), None) if not conn: await api.edit_message(chat_id, message_id, f" {_tt(lang, 'connection_not_found')}") return conn_name = conn.get("name") or name or "Connection" rows = [ [{"text": f"🗑 {_tt(lang, 'btn_delete')}", "callback_data": _ref("user_delete_confirm", {"conn_id": conn_id})}], [{"text": f"❌ {_tt(lang, 'btn_cancel')}", "callback_data": "refresh"}], ] await api.edit_message( chat_id, message_id, _tt(lang, "delete_connection_confirm", name=_e(conn_name)) + f"\n\n{_tt(lang, 'cannot_be_undone')}", reply_markup={"inline_keyboard": rows}, ) async def _user_delete_confirm(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, tg_id: str, tg_username: Optional[str], conn_id: str, load_data_fn: Callable, self_service_svc, lang: str = "en"): panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.answer_callback(callback_id, text="Access denied") await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'access_denied')}") return if not self_service_svc: await api.answer_callback(callback_id, text="Self-service not available") await api.edit_message(chat_id, message_id, _tt(lang, "self_service_unavailable")) return await api.answer_callback(callback_id) data = load_data_fn() if not _self_service_telegram_enabled(data): await api.edit_message(chat_id, message_id, _tt(lang, "self_service_disabled_contact")) return try: await self_service_svc.delete_user_connection(panel_user["id"], conn_id, "telegram") except Exception as e: logger.exception("Bot: self-service delete failed") await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'error')}: {_tt(lang, 'self_service_unavailable')}") return data = load_data_fn() conns = [c for c in data.get("user_connections", []) if c.get("user_id") == panel_user.get("id")] if not conns: if _self_service_telegram_enabled(data): kb = _build_connections_keyboard(conns, data, lang) await api.edit_message(chat_id, message_id, f"✅ {_tt(lang, 'connection_deleted')}\n\n{_tt(lang, 'no_connections_create_short')}", reply_markup=kb) else: await api.edit_message(chat_id, message_id, f"✅ {_tt(lang, 'connection_deleted_no_connections')}") return kb = _build_connections_keyboard(conns, data, lang) await api.edit_message(chat_id, message_id, f"✅ {_tt(lang, 'connection_deleted')}\n\n{_tt(lang, 'your_connections', count=len(conns))}", reply_markup=kb) async def _user_add_client_final(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, tg_id: str, tg_username: Optional[str], server_id: int, proto: str, name: str, load_data_fn: Callable, generate_vpn_link_fn: Callable, self_service_svc, lang: str = "en"): panel_user = _find_user(load_data_fn, tg_id, tg_username) if not panel_user: await api.answer_callback(callback_id, text="Access denied") await api.edit_message(chat_id, message_id, f"❌ {_tt(lang, 'access_denied')}") return if not self_service_svc: await api.answer_callback(callback_id, text="Self-service not available") await api.edit_message(chat_id, message_id, _tt(lang, "self_service_unavailable")) return data = load_data_fn() if not _self_service_telegram_enabled(data): await api.answer_callback(callback_id) await api.edit_message(chat_id, message_id, _tt(lang, "self_service_disabled_contact")) return await api.answer_callback(callback_id) clean_name = (name or "Connection").strip()[:80] await _user_create_connection(api, chat_id, panel_user, server_id, proto, clean_name, load_data_fn, generate_vpn_link_fn, self_service_svc, lang) async def _user_create_connection(api: TelegramAPI, chat_id: int, panel_user: dict, server_id: int, proto: str, name: str, load_data_fn: Callable, generate_vpn_link_fn: Callable, self_service_svc, lang: str = "en"): try: result = await self_service_svc.create_user_connection(panel_user["id"], server_id, proto, name, "telegram") loading_msg = await api.send_message(chat_id, _tt(lang, "creating_connection_named", name=_e(name))) loading_msg_id = loading_msg.get("result", {}).get("message_id") data = load_data_fn() servers = data.get("servers", []) if server_id < 0 or server_id >= len(servers): await api.edit_message(chat_id, loading_msg_id, f"❌ {_tt(lang, 'server_not_found')}") return server = servers[server_id] config = result.get("config", "") if config: if loading_msg_id: await api.call("deleteMessage", chat_id=chat_id, message_id=loading_msg_id) await _send_config_text(api, chat_id, server, proto, name, config, generate_vpn_link_fn, lang) else: conn = result.get("connection", {}) client_id = conn.get("client_id") if client_id: if loading_msg_id: await api.call("deleteMessage", chat_id=chat_id, message_id=loading_msg_id) await _send_config_by_client(api, chat_id, server, proto, client_id, name, generate_vpn_link_fn, lang) else: if loading_msg_id: await api.edit_message(chat_id, loading_msg_id, f"✅ {_tt(lang, 'connection_created_success')}") except Exception as e: logger.exception("Bot: self-service create failed") await api.send_message(chat_id, f"❌ {_tt(lang, 'error')}: {_tt(lang, 'self_service_unavailable')}") # ----------------------------------------------------------------------- # # Main polling loop and dispatcher # ----------------------------------------------------------------------- # async def _run_bot(token: str, load_data_fn: Callable, generate_vpn_link_fn: Callable, save_data_fn: Optional[Callable] = None, self_service_svc=None): offset = 0 logger.info("Telegram bot started (raw httpx polling).") async with httpx.AsyncClient() as client: api = TelegramAPI(token, client) me = await api.call("getMe") if not me.get("ok"): logger.error(f"Telegram bot: invalid token or API error: {me}") return logger.info(f"Telegram bot logged in as @{me['result']['username']}") await _set_default_commands(api) while True: try: updates = await api.get_updates(offset=offset, timeout=25) except asyncio.CancelledError: logger.info("Telegram bot polling cancelled.") return except Exception as e: logger.warning(f"Telegram bot polling error: {e}") await asyncio.sleep(5) continue for update in updates: offset = update["update_id"] + 1 try: await _dispatch(api, update, load_data_fn, generate_vpn_link_fn, save_data_fn, self_service_svc) except asyncio.CancelledError: return except Exception as e: logger.exception(f"Telegram bot: error handling update {update['update_id']}: {e}") async def _dispatch(api: TelegramAPI, update: dict, load_data_fn: Callable, generate_vpn_link_fn: Callable, save_data_fn: Optional[Callable] = None, self_service_svc=None): if "message" in update: msg = update["message"] text = msg.get("text", "") tg_username = msg["from"].get("username") lang = _tg_lang(msg["from"]) chat_id = msg["chat"]["id"] if text.startswith(("/connections", "/connect", "/disconnect")) and not _is_private_chat(msg.get("chat")): await _reject_non_private(api, chat_id, lang) return if await _handle_pending_input(api, msg, load_data_fn, save_data_fn, generate_vpn_link_fn, self_service_svc): return if text.startswith("/start") or text.startswith("/admin"): await _handle_start(api, msg, load_data_fn) elif text.startswith("/connections"): panel_user = _find_user(load_data_fn, str(msg["from"]["id"]), tg_username) if not panel_user: await api.send_message(chat_id, f" {_tt(lang, 'access_denied')}") else: await _send_user_connections(api, chat_id, panel_user, load_data_fn, lang=lang) elif text.startswith("/connect"): await _user_create_start_message(api, chat_id, str(msg["from"]["id"]), tg_username, load_data_fn, lang) elif text.startswith("/disconnect"): panel_user = _find_user(load_data_fn, str(msg["from"]["id"]), tg_username) if not panel_user: await api.send_message(chat_id, f" {_tt(lang, 'access_denied')}") else: await _send_user_connections(api, chat_id, panel_user, load_data_fn, lang=lang) elif text.startswith("/servers"): if _require_admin(load_data_fn, str(msg["from"]["id"]), tg_username): await _admin_servers(api, msg["chat"]["id"], None, load_data_fn, lang) else: await api.send_message(msg["chat"]["id"], f" {_tt(lang, 'access_denied')}") elif text.startswith("/addserver"): await _handle_add_server_command(api, msg, load_data_fn, save_data_fn, lang) elif "callback_query" in update: cq = update["callback_query"] callback_id = cq["id"] data_str = cq.get("data", "") chat_id = cq["message"]["chat"]["id"] chat = cq["message"].get("chat", {}) message_id = cq["message"]["message_id"] tg_id = str(cq["from"]["id"]) tg_username = cq["from"].get("username") lang = _tg_lang(cq["from"]) if data_str == "noop": await api.answer_callback(callback_id) return if data_str == "refresh": await _handle_refresh(api, chat_id, message_id, callback_id, tg_id, tg_username, load_data_fn, lang) return if data_str.startswith("cfg:"): if not _is_private_chat(chat): await _reject_non_private(api, chat_id, lang, callback_id) return await _handle_get_config(api, chat_id, message_id, callback_id, data_str[4:], tg_id, tg_username, load_data_fn, generate_vpn_link_fn, lang) return # Self-service user callbacks (before admin gate). if data_str == "user_create": if not _is_private_chat(chat): await _reject_non_private(api, chat_id, lang, callback_id) return await _user_create_start(api, chat_id, message_id, callback_id, tg_id, tg_username, load_data_fn, self_service_svc, lang) return if data_str == "user_create_cancel": await _user_create_cancel(api, chat_id, message_id, callback_id, tg_id, tg_username, load_data_fn, lang) return ref = _resolve_ref(data_str) if ref: action = ref.get("action") payload = ref.get("payload", {}) if action == "user_create_server": if not _is_private_chat(chat): await _reject_non_private(api, chat_id, lang, callback_id) return await _user_create_server(api, chat_id, message_id, callback_id, tg_id, tg_username, int(payload.get("sid", 0)), load_data_fn, self_service_svc, lang) return if action == "user_create_protocol": if not _is_private_chat(chat): await _reject_non_private(api, chat_id, lang, callback_id) return await _user_create_protocol(api, chat_id, message_id, callback_id, tg_id, tg_username, int(payload.get("sid", 0)), payload.get("proto", "awg"), load_data_fn, lang) return if action == "user_add_client": if not _is_private_chat(chat): await _reject_non_private(api, chat_id, lang, callback_id) return await _user_add_client_final(api, chat_id, message_id, callback_id, tg_id, tg_username, int(payload.get("sid", 0)), payload.get("proto", "awg"), payload.get("name", ""), load_data_fn, generate_vpn_link_fn, self_service_svc, lang) return if action == "user_delete": if not _is_private_chat(chat): await _reject_non_private(api, chat_id, lang, callback_id) return await _user_delete(api, chat_id, message_id, callback_id, tg_id, tg_username, payload.get("conn_id", ""), payload.get("name", ""), load_data_fn, lang) return if action == "user_delete_confirm": if not _is_private_chat(chat): await _reject_non_private(api, chat_id, lang, callback_id) return await _user_delete_confirm(api, chat_id, message_id, callback_id, tg_id, tg_username, payload.get("conn_id", ""), load_data_fn, self_service_svc, lang) return panel_user = _require_admin(load_data_fn, tg_id, tg_username) if not panel_user: await api.answer_callback(callback_id, "Access denied") return await api.answer_callback(callback_id) if data_str == "adm:menu": await api.edit_message(chat_id, message_id, f"{_tt(lang, 'admin_menu')}", reply_markup=_admin_main_keyboard(lang)) elif data_str == "adm:servers": await _admin_servers(api, chat_id, message_id, load_data_fn, lang) elif data_str == "adm:users": await _admin_users(api, chat_id, message_id, load_data_fn, lang) elif data_str == "adm:myconns": await _send_user_connections(api, chat_id, panel_user, load_data_fn, lang=lang) elif data_str == "adm:addserver_help": await api.edit_message( chat_id, message_id, f"➕ {_tt(lang, 'add_server_label')}\n\n" f"{_tt(lang, 'add_server_command')}\n", reply_markup={"inline_keyboard": [[{"text": f"⬅️ {_tt(lang, 'btn_admin_menu')}", "callback_data": "adm:menu"}]]}, ) elif data_str.startswith("srv:"): await _admin_server_detail(api, chat_id, message_id, int(data_str.split(":", 1)[1]), load_data_fn, lang) else: ref = _resolve_ref(data_str) if not ref: await api.edit_message(chat_id, message_id, f" {_tt(lang, 'action_expired')}") return action = ref.get("action") payload = ref.get("payload", {}) sid = int(payload.get("sid", 0) or 0) proto = payload.get("proto", "awg") if action == "user": await _admin_user_detail(api, chat_id, message_id, payload.get("uid"), load_data_fn, lang) elif action == "proto": await _admin_protocol_detail(api, chat_id, message_id, sid, proto, load_data_fn, lang) elif action == "toggle_proto": await _admin_toggle_protocol(api, chat_id, message_id, sid, proto, bool(payload.get("start")), load_data_fn, lang) elif action == "clients": await _admin_clients(api, chat_id, message_id, sid, proto, load_data_fn, lang) elif action == "client": await _admin_client_detail(api, chat_id, message_id, sid, proto, payload.get("client", {}), lang) elif action == "client_cfg": if not _is_private_chat(chat): await _reject_non_private(api, chat_id, lang, callback_id) return data = load_data_fn() server = data["servers"][sid] await _send_config_by_client(api, chat_id, server, proto, payload.get("client_id"), payload.get("name", "Connection"), generate_vpn_link_fn, lang) elif action == "add_client": await _admin_add_client(api, chat_id, message_id, sid, proto, panel_user, load_data_fn, save_data_fn, generate_vpn_link_fn, lang, tg_id) elif action == "create_client": await _admin_create_client(api, chat_id, message_id, sid, proto, payload.get("name", "Connection"), payload.get("user_id"), load_data_fn, save_data_fn, generate_vpn_link_fn, lang) elif action == "toggle_client": await _admin_toggle_client(api, chat_id, message_id, sid, proto, payload.get("client_id"), bool(payload.get("enable")), load_data_fn, lang) elif action == "remove_client": await _admin_remove_client(api, chat_id, message_id, sid, proto, payload.get("client_id"), load_data_fn, save_data_fn, lang)