"""A native GusVoice UI for an already installed AWG Toolza server.""" from __future__ import annotations import asyncio import contextlib from dataclasses import dataclass import logging from pathlib import Path import re import secrets import shutil import time from gusvoice_bot import GusVoiceApiError, InputFile from . import __version__ from .core import AwgError, EXPIRE, LIMIT, NAME, SPEED, esc, redacted, size log = logging.getLogger("awg-gus") PAGE = 10 FILE_MAX = 25 * 1024 * 1024 def button(text, data): if len(data.encode()) > 64: raise ValueError("callback_data too long") return {"text": text, "callback_data": data} HOME = [button("🏠 Меню", "home")] def client_status(c): if c.get("blocked"): return "🚫 заблокирован" if c.get("online"): return "🟢 онлайн" return "⚪ офлайн" if c.get("handshake") else "⚪ ещё не подключался" @dataclass class Event: uid: str chat: str update_id: int message: str | None = None text: str = "" data: str = "" class Bot: def __init__(self, config, api, delivery, state): self.config, self.api, self.io, self.state = config, api, delivery, state self.owners = set(config["owners"]) self.steps = {} self.confirmations = {} self.tasks = {} self.interrupted = {row[0] for row in state.db.execute("SELECT uid FROM pending")} with state.db: state.db.execute("DELETE FROM pending") self.observed = {} async def screen(self, e, text, rows=None): out = await self.io.show(e.chat, text, rows if rows is not None else [HOME], e.message) if isinstance(out, dict): e.message = out.get("message_id") return out async def clients(self): data = (await self.api.call("clients", "list")).get("data") if not isinstance(data, list): raise AwgError("Неверный ответ clients list") return [c for c in data if isinstance(c, dict) and isinstance(c.get("name"), str) and NAME.fullmatch(c["name"])] async def client(self, name): if not NAME.fullmatch(name): raise ValueError("Неверное имя клиента") c = next((c for c in await self.clients() if c["name"] == name), None) if c is None: raise ValueError("Клиент уже удалён. Обновите список.") return c async def process(self, update): n = update.get("update_id") if not isinstance(n, int) or isinstance(n, bool) or n < self.state.get("offset", 0): return cb = update.get("callback_query") msg = cb.get("message") if isinstance(cb, dict) else update.get("message") author = cb.get("from") if isinstance(cb, dict) else (msg or {}).get("from") msg, author = msg or {}, author or {} chat = msg.get("chat") or {} uid, cid = author.get("id"), chat.get("id") allowed = (chat.get("type") == "dm" and uid in self.owners and isinstance(cid, str)) if not allowed: self.state.claim(n) if isinstance(cb, dict): await self.io.ack(cb["id"], "Доступ только владельцу в личных сообщениях") elif chat.get("type") == "dm" and (msg.get("text") or "").strip() == "/whoami": await self.io.show(cid, f"Ваш ID: {esc(uid)}\nДоступ к управлению не выдан.") return key = (uid, cid) busy = key in self.tasks and not self.tasks[key].done() # Persist deduplication before ack or starting a privileged operation. self.state.claim(n, uid, cid) if isinstance(cb, dict): try: await self.io.ack(cb["id"], "Предыдущая операция ещё выполняется" if busy else None) except Exception: self.state.finish(n) raise if busy: self.state.finish(n) if not cb: await self.io.show(cid, "⏳ Предыдущая операция ещё выполняется. Дождитесь результата.") return e = Event(uid, cid, n, msg.get("message_id") if cb else None, msg.get("text") or "", cb.get("data") or "" if cb else "") self.tasks[key] = asyncio.create_task(self._handle(e), name="awg-gus-action") async def _handle(self, e): completed = True try: if e.uid in self.interrupted: self.interrupted.remove(e.uid) await self.io.show(e.chat, "⚠️ Предыдущий запуск прервался во время обработки. " "Проверьте состояние: незавершённые действия автоматически не повторяются.") if e.data: await self.action(e) else: await self.message(e) except asyncio.CancelledError: completed = False raise except (AwgError, ValueError) as exc: await self._error(e, str(exc)) except Exception as exc: # API/transport exceptions may contain a sensitive URL. Log class only. log.error("Handler failed: %s (update %s)", type(exc).__name__, e.update_id) await self._error(e, "Ошибка связи или обработки. Если действие изменяло сервер, проверьте результат перед повтором.") finally: if completed: self.state.finish(e.update_id) def cancel_confirmations(self, e): self.confirmations = {k: v for k, v in self.confirmations.items() if v[:2] != (e.uid, e.chat)} async def _error(self, e, text): with contextlib.suppress(Exception): await self.screen(e, "❌ " + esc(redacted(text)[:650])) async def message(self, e): text = e.text.strip() command = text.split(maxsplit=1)[0].split("@", 1)[0] if text else "" if command.startswith("/"): self.cancel_confirmations(e) if command in {"/start", "/menu", "/status", "/cancel"}: self.steps.pop((e.uid, e.chat), None) await self.home(e) elif command == "/clients": await self.list_clients(e) elif command == "/whoami": await self.screen(e, f"Ваш ID: {esc(e.uid)}\nДоступ: владелец.") elif command == "/help": await self.screen(e, "AWG для Гуся\n/menu — главное меню\n/clients — клиенты\n" "/cancel — отменить ввод\n/whoami — ваш ID\n\n" "Изменения клиентов общие с Telegram-ботом: оба управляют одним awg2.") elif command == "/new": await self.prompt(e, "add", "", "Имя нового клиента: латиница, цифры, _ и -, до 32 символов.") elif (e.uid, e.chat) in self.steps: await self.input(e, text) else: await self.home(e) async def home(self, e): d = (await self.api.call("status")).get("data") or {} s = d.get("server") or {} state = "🟢 работает" if s.get("up") else "🔴 не поднят" text = (f"🪿 AWG для Гуся · {__version__}\n" + (f"Подключение: {esc(self.config['label'])}\n" if self.config.get("label") else "") + f"Toolza: {esc(d.get('version', '?'))}\n" f"Сервер: {esc(d.get('host', '?'))}\nAWG: {state}\n" f"Клиенты: {s.get('clients', 0)} · онлайн {s.get('online', 0)}\n" f"Протокол: {esc(s.get('proto', '?'))} · порт {esc(s.get('port', '?'))}") await self.screen(e, text, [ [button("👥 Клиенты", "list:0"), button("➕ Создать", "new")], [button("🖥 Сервер", "server"), button("🩺 Диагностика", "diag")], [button("💾 Резервная копия", "backup"), button("📦 Экспорт клиентов", "export")], [button("🔔 Уведомления", "notices"), button("ℹ️ О боте", "about")], [button("🔄 Обновить", "home")]]) async def list_clients(self, e, page=0): rows = sorted(await self.clients(), key=lambda c: (not c.get("online"), c["name"].lower())) page = min(max(0, page), max(0, (len(rows) - 1) // PAGE)) buttons = [[button(f"{'🚫' if c.get('blocked') else '🟢' if c.get('online') else '⚪'} {c['name']}", "client:" + c["name"])] for c in rows[page * PAGE:(page + 1) * PAGE]] nav = [] if page: nav.append(button("←", f"list:{page-1}")) if (page + 1) * PAGE < len(rows): nav.append(button("→", f"list:{page+1}")) if nav: buttons.append(nav) buttons += [[button("➕ Создать", "new"), button("🔄 Обновить", f"list:{page}")], HOME] await self.screen(e, f"👥 Клиенты: {len(rows)}\nСтраница {page+1}. Выберите клиента.", buttons) async def card(self, e, name): c = await self.client(name) exp = c.get("expires") expiry = time.strftime("%Y-%m-%d %H:%M", time.localtime(exp)) if exp else "без ограничения" text = (f"👤 {esc(name)}\n{client_status(c)}\nIP: {esc(c.get('ip', ''))}\n" f"Трафик: ↓ {size(c.get('rx'))} · ↑ {size(c.get('tx'))}\nСрок: {expiry} (время сервера)\n") text += (f"Лимит: {size(c.get('used'))} / {size(c['limit'])} ({esc(c.get('period', 'month'))})\n" if c.get("limit") else "Лимит: нет\n") text += f"Скорость: {esc(c.get('speed') or 'не ограничена')}" + (" Мбит/с" if c.get("speed") else "") watching = self.state.watched(e.uid, name) await self.screen(e, text, [ [button("📄 Конфиг + QR", f"conf:{name}"), button("🔄 Обновить", f"client:{name}")], [button("📅 Срок", f"expiry:{name}"), button("📊 Лимит", f"limit:{name}")], [button("🚀 Скорость", f"speed:{name}"), button("✏️ Имя", f"rename:{name}")], [button("🔑 Новый ключ", f"reissue:{name}"), button("🗑 Удалить", f"delete:{name}")], [button("🔕 Не следить" if watching else "🔔 Следить", f"watch:{name}"), button("♻️ Сбросить трафик", f"reset:{name}")], [button("← Клиенты", "list:0")], HOME]) async def action(self, e): op, _, arg = e.data.partition(":") if op != "yes": self.steps.pop((e.uid, e.chat), None) self.cancel_confirmations(e) if op == "home": await self.home(e) elif op == "list": await self.list_clients(e, int(arg) if arg.isdigit() else 0) elif op == "new": await self.prompt(e, "add", "", "Имя нового клиента: латиница, цифры, _ и -, до 32 символов.") elif op == "client": await self.card(e, arg) elif op == "conf": await self.client(arg) await self.send_config(e, arg) elif op == "expiry": await self.client(arg) await self.screen(e, f"Срок: {esc(arg)}\nОтсчитывается от момента применения.", [ [button("1 день", f"exp:{arg}:+1d"), button("7 дней", f"exp:{arg}:+7d")], [button("30 дней", f"exp:{arg}:+30d"), button("Без срока", f"exp:{arg}:off")], [button("← Клиент", f"client:{arg}")]]) elif op == "exp": name, _, value = arg.partition(":") await self.client(name) args = ["client", "unexpire", name] if value == "off" else ["client", "expire", name, value] await self.confirm(e, f"Изменить срок {esc(name)}: {esc(value)}?", args, name) elif op in {"rename", "limit", "speed"}: await self.client(arg) prompt = {"rename": "Новое имя: латиница, цифры, _ и -, до 32 символов.", "limit": "Лимит: например 50G или 500M. Добавьте total для общего лимита: 50G total.\n" "По умолчанию — за месяц. off — снять ограничение.", "speed": "Скорость в Мбит/с: целое число, например 20. off — снять ограничение."}[op] await self.prompt(e, op, arg, prompt) elif op in {"delete", "reissue", "reset"}: await self.client(arg) verb, label = {"delete": ("del", "Удалить клиента и его доступ"), "reissue": ("reissue", "Перевыпустить ключ: старый конфиг перестанет работать"), "reset": ("limit-reset", "Сбросить счётчик лимита трафика")}[op] await self.confirm(e, f"{label}: {esc(arg)}?", ["client", verb, arg], arg) elif op == "watch": await self.client(arg) self.state.watch(e.uid, arg) await self.card(e, arg) elif op == "newexp": draft = self.state.get(f"draft:{e.uid}:{e.chat}") if not draft or time.time() - draft["time"] > 600: raise ValueError("Ввод устарел. Создайте клиента заново.") value = arg args = ["client", "add", draft["name"], "mimicry=server"] if value != "off": args.append("expire=" + value) await self.confirm(e, f"Создать {esc(draft['name'])}?\nСрок: {esc(value)}; мимикрия сервера.", args, draft["name"]) elif op == "yes": await self.execute(e, arg) elif op == "server": d = (await self.api.call("server", "info")).get("data") or {} text = "🖥 Сервер AWG\n" + "\n".join( f"{label}: {esc(d.get(key, '—'))}" for key, label in [("up", "Работает"), ("proto", "Протокол"), ("profile", "Профиль"), ("endpoint", "Endpoint"), ("net", "Подсеть"), ("mtu", "MTU"), ("clients", "Клиенты")]) await self.screen(e, text, [[button("🔄 Перезапустить AWG", "restart")], HOME]) elif op == "restart": await self.confirm(e, "Перезапустить интерфейс AWG? Соединения VPN кратковременно прервутся.", ["server", "restart"]) elif op == "diag": r = await self.api.call("diag", "status") text = redacted(r.get("log") or "Диагностика завершена") text = re.sub(r"\x1b\[[0-9;]*[A-Za-z]", "", text) await self.screen(e, "🩺 Диагностика\n
" + esc(text[-1800:]) + "
") elif op in {"backup", "export"}: args = ["backup", "create"] if op == "backup" else ["clients", "export"] await self.confirm(e, "Создать и прислать архив в эту личную беседу? Архив содержит ключи VPN.", args) elif op == "notices": on = self.state.get(f"notify:{e.uid}", False) watches = [r[0] for r in self.state.db.execute("SELECT name FROM watches WHERE uid=?", (e.uid,))] await self.screen(e, "🔔 Уведомления\nСостояние сервера: " + ("включены" if on else "выключены") + "\nНаблюдение за клиентами: " + esc(", ".join(watches) or "нет") + "\nПроверка раз в минуту. Клиент офлайн после 5 минут без рукопожатия.\n" "При старте бот запоминает состояние без рассылки.", [[button("Выключить серверные" if on else "Включить серверные", "notify")], HOME]) elif op == "notify": key = f"notify:{e.uid}" self.state.set(key, not self.state.get(key, False)) e.data = "notices" await self.action(e) elif op == "about": await self.screen(e, f"AWG для Гуся {__version__}\n" "Отдельная служба: awg-gus-bot. Управление тем же awg2, что и в Telegram.\n" "Обновление на сервере: sudo awg-gus-bot update\n" "Проверка: sudo awg-gus-bot check\n" "В этой версии: клиенты, ключи/QR, сроки, трафик, скорость, диагностика, " "бэкап и экспорт, перезапуск AWG, уведомления.\n" "Установка AWG, туннели, обновление ядра и восстановление бэкапа — через меню awg2.") else: raise ValueError("Кнопка устарела. Откройте /menu.") async def prompt(self, e, kind, name, text): self.steps[(e.uid, e.chat)] = (kind, name, time.monotonic() + 600) await self.screen(e, text + "\n\nОтмена: /cancel", [HOME]) async def input(self, e, text): key = (e.uid, e.chat) kind, name, expires = self.steps[key] if time.monotonic() > expires: self.steps.pop(key, None) raise ValueError("Ввод устарел. Откройте меню заново.") if kind in {"add", "rename"}: if not NAME.fullmatch(text): raise ValueError("Имя: латиница, цифры, _ и -, от 1 до 32 символов. Повторите ввод.") if any(c["name"] == text for c in await self.clients()): raise ValueError("Такое имя уже существует. Введите другое.") if kind == "add": self.state.set(f"draft:{e.uid}:{e.chat}", {"name": text, "time": time.time()}) await self.screen(e, f"Срок нового клиента {esc(text)}:", [ [button("1 день", "newexp:+1d"), button("7 дней", "newexp:+7d")], [button("30 дней", "newexp:+30d"), button("Без срока", "newexp:off")], HOME]) else: await self.confirm(e, f"Переименовать {esc(name)} → {esc(text)}?", ["client", "rename", name, text], text) elif kind == "limit": parts = text.split() value = parts[0] if parts else "" value = value if value == "off" else value.upper() period = parts[1] if len(parts) == 2 else "month" if len(parts) > 2 or not LIMIT.fullmatch(value) or period not in {"month", "total"}: raise ValueError("Введите 50G, 500M, 50G total или off.") await self.confirm(e, f"Лимит {esc(name)}: {esc(value)} ({period})?", ["client", "limit", name, value, period], name) elif kind == "speed": if not SPEED.fullmatch(text): raise ValueError("Введите целое число Мбит/с или off.") await self.confirm(e, f"Скорость {esc(name)}: {esc(text)} Мбит/с?", ["client", "speed", name, text], name) self.steps.pop(key, None) async def confirm(self, e, text, args, name=""): from .core import validate_command validate_command(args) now = time.monotonic() self.confirmations = {k: v for k, v in self.confirmations.items() if v[0:2] != (e.uid, e.chat) and v[2] > now} token = secrets.token_hex(8) self.confirmations[token] = (e.uid, e.chat, now + 120, args, name) await self.screen(e, text + "\n\nПодтверждение действует 2 минуты.", [[button("✅ Подтвердить", "yes:" + token), button("Отмена", "home")]]) async def execute(self, e, token): pending = self.confirmations.get(token) if not pending or pending[:2] != (e.uid, e.chat) or pending[2] < time.monotonic(): raise ValueError("Подтверждение устарело или уже использовано. Повторите действие из меню.") self.confirmations.pop(token) args, name = pending[3:] await self.screen(e, "⏳ Выполняю…") r = await self.api.call(*args) await self.screen(e, "✅ Выполнено", [[button("👥 Клиенты", "list:0")], HOME]) if args[:2] in (["client", "add"], ["client", "reissue"]): await self.send_config(e, name) elif args[:2] in (["backup", "create"], ["clients", "export"]): data = r.get("data") or {} path = Path(data.get("path") or data.get("file") or "") if not path.is_absolute() or not path.is_file(): raise AwgError("Архив создан, но awg2 не вернул путь к файлу.") if path.stat().st_size > FILE_MAX: await self.io.show(e.chat, "Архив больше 25 МиБ и сохранён на сервере. Заберите его через SSH.") else: await self.io.call("send_message", chat_id=e.chat, text="Архив AWG: содержит ключи VPN.", files=[InputFile(path)]) async def send_config(self, e, name): data = (await self.api.call("client", "conf", name)).get("data") or {} text = data.get("text") if not isinstance(text, str) or not text or len(text.encode()) > FILE_MAX: raise AwgError("Конфиг отсутствует или превышает допустимый размер.") files = [InputFile(text.encode(), filename=name + ".conf", content_type="text/plain")] note = "" if len(text.encode()) <= 2800 and shutil.which("qrencode"): proc = await asyncio.create_subprocess_exec("qrencode", "-t", "PNG", "-o", "-", "-s", "8", "-m", "2", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL) try: png, _ = await asyncio.wait_for(proc.communicate(text.encode()), 10) if proc.returncode == 0 and png.startswith(b"\x89PNG\r\n\x1a\n"): files.append(InputFile(png, filename=name + ".png", content_type="image/png")) else: note = " QR не сформирован; импортируйте файл." except (asyncio.TimeoutError, asyncio.CancelledError) as exc: with contextlib.suppress(ProcessLookupError): proc.kill() await proc.wait() if isinstance(exc, asyncio.CancelledError): raise note = " QR не сформирован; импортируйте файл." else: note = " QR недоступен или конфиг слишком длинный; импортируйте файл." await self.io.call("send_message", chat_id=e.chat, text=f"{name} — импорт в AmneziaVPN / AmneziaWG.{note}", files=files) async def monitor_tick(self): watches = list(self.state.db.execute("SELECT uid,name FROM watches")) notify = [uid for uid in self.owners if self.state.get(f"notify:{uid}", False)] if not watches and not notify: return clients = {c["name"]: c for c in await self.clients()} if watches else {} current = {} for uid, name in watches: if uid not in self.owners: continue c = clients.get(name) status = "удалён" if c is None else "заблокирован" if c.get("blocked") else ( "онлайн" if c.get("handshake") and float(c.get("ago") or 0) < 300 else "офлайн") current[(uid, name)] = status if notify: d = (await self.api.call("server", "info")).get("data") or {} for uid in notify: current[(uid, "Сервер AWG")] = "работает" if d.get("up") else "не поднят" for key, status in current.items(): previous = self.observed.get(key) if previous is not None and previous != status: row = self.state.db.execute("SELECT chat FROM chats WHERE uid=?", (key[0],)).fetchone() if row: await self.io.show(row[0], f"🔔 {esc(key[1])}: {esc(previous)} → {esc(status)}") self.observed = current async def monitor(self): while True: try: await self.monitor_tick() except Exception as exc: log.warning("Monitor failed: %s", type(exc).__name__) await asyncio.sleep(60) async def run(self): webhook = await self.io.call("get_webhook_info") if webhook.get("url"): raise ValueError("У этого бота уже настроен webhook. Используйте отдельного бота Гуся.") await self.io.call("set_my_commands", commands=[ {"command": "menu", "description": "Управление AWG"}, {"command": "clients", "description": "Клиенты VPN"}, {"command": "new", "description": "Создать клиента"}, {"command": "cancel", "description": "Отменить ввод"}, {"command": "help", "description": "Помощь"}]) monitor = asyncio.create_task(self.monitor()) delay = 1 try: while True: try: updates = await self.io.call("get_updates", offset=self.state.get("offset", 0), timeout=30, allowed_updates=["message", "callback_query"]) for update in updates: await self.process(update) delay = 1 except GusVoiceApiError as exc: if exc.code in {401, 403, 404, 409}: log.error("Polling stopped: HTTP %s. Check token, enabled bots, webhook and duplicate processes.", exc.code) raise log.warning("Polling retry: HTTP %s", exc.code) await asyncio.sleep(delay) delay = min(30, delay * 2) except (OSError, TimeoutError) as exc: log.warning("Network retry: %s", type(exc).__name__) await asyncio.sleep(delay) delay = min(30, delay * 2) finally: monitor.cancel() for task in self.tasks.values(): task.cancel() await asyncio.gather(monitor, *self.tasks.values(), return_exceptions=True)