kobalt/swiftgram
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

new file: README.md

Browse Source 
modified:   bot.py
This commit is contained in:
kobaltgit
2026-04-05 22:16:01 +03:00
parent 9482cfe7f6
commit 1747e8e969
2 changed files with 216 additions and 452 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
README.md Normal file
View File

@@ -0,0 +1,72 @@
# 🚀 SwiftGram MTProxy
**SwiftGram** — это интеллектуальный и чистый менеджер MTProxy для Telegram, ориентированный на скорость, стабильность и отсутствие рекламы. Скрипт автоматически оптимизирует сетевой стек сервера и обеспечивает работу звонков через прокси.
---
## ✨ Ключевые особенности
* **🚫 Без рекламы:** Полностью удалены промокоды, донаты и рекламные ссылки. Только чистый код.
* **📡 Поддержка IPv4 + IPv6:** Прокси автоматически слушает оба протокола. Ссылки для подключения генерируются для обоих типов адресов.
* **📞 Исправленные звонки (UDP):** Автоматическая настройка Firewall (UFW/Firewalld) и проброс UDP-портов в Docker для работы голосовых и видеовызовов.
* **🏎 Оптимизация BBR:** Включение алгоритма контроля перегрузки Google BBR на уровне ядра для минимального пинга и максимальной скорости.
* **🧠 Умный выбор порта:** Скрипт проверяет доступность порта 443. Если он занят (например, панелью Hiddify или Nginx), SwiftGram предложит свободный альтернативный порт, не нарушая работу других служб.
* **🔍 Анализ маскировки:** Перед установкой скрипт анализирует задержку (ping) до популярных доменов и выбирает лучший вариант для Fake TLS.
* **🤖 Управление через Bot:** Полноценный Telegram-бот для мониторинга статуса, получения ссылок, чтения логов и перезагрузки контейнера.
---
## 🚀 Быстрая установка
Выполните одну команду в терминале вашего сервера (Ubuntu/Debian/CentOS):
```bash
curl -sL https://git.bargcraft.top/kobalt/swiftgram/raw/branch/main/install.sh -o /usr/local/bin/swiftgram && chmod +x /usr/local/bin/swiftgram && swiftgram
```
После установки вы сможете запускать менеджер просто командой: `swiftgram`
---
## 🤖 Настройка Telegram-бота
Чтобы управлять прокси прямо из Telegram:
1. Создайте нового бота у [@BotFather](https://t.me/BotFather) и получите **API Token**.
2. Узнайте свой Telegram ID (через @userinfobot или аналоги).
3. В меню `swiftgram` выберите пункт **3 (Настроить Telegram-бот)**.
4. Введите токен и ваш ID.
**Доступные команды бота:**
* `/status` — Детальная диагностика (BBR, IPv6, UDP, порт).
* `/link` — Получение ссылок tg://proxy.
* `/share` — Красивое сообщение с данными прокси для друзей.
* `/restart` — Перезагрузка Docker-контейнера.
* `/logs` — Просмотр последних 30 строк логов.
* `/remove` — Удаление прокси с сервера.
---
## 🛠 Техническая информация
* **Контейнер:** [nineseconds/mtg:2](https://github.com/9seconds/mtg)
* **Путь установки:** `/opt/swiftgram`
* **Сетевой режим:** Docker Bridge с пробросом TCP+UDP.
* **Firewall:** Скрипт автоматически открывает выбранный порт в `ufw` или `firewalld`.
---
## 🗑 Удаление
SwiftGram поддерживает полную очистку системы. Выберите пункт **5** в главном меню или введите в консоли:
`swiftgram` -> пункт 5.
Скрипт удалит контейнер, системный сервис бота, все файлы конфигурации и самого себя.
---
## 🛡 Безопасность
Проект SwiftGram является приватным инструментом. Мы рекомендуем использовать сложные Secret-ключи и ограничивать список `ALLOWED_IDS` в настройках бота, чтобы посторонние не могли управлять вашим сервером.
---
**Разработано для SwiftGram Community.**

582
bot.py
View File

@@ -1,8 +1,8 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
GoTelegram MTProxy — Telegram-бот для управления MTProxy на сервере. SwiftGram MTProxy — Telegram-бот для управления MTProxy на сервере.
Кнопочное меню, проверка портов, совместимость с Amnezia/3x-ui, Возможности: статус BBR, проверка IPv6, фикс звонков (UDP), управление контейнером.
кнопка «Поделиться ключом», TCP+UDP для звонков. Без рекламы и промокодов.
""" """
import asyncio import asyncio
@@ -12,9 +12,11 @@ import os
import re import re
from pathlib import Path from pathlib import Path
# ── Загрузка настроек из .env ────────────────────────────────────────────────
_env_path = Path(__file__).resolve().parent / ".env" _env_path = Path(__file__).resolve().parent / ".env"
if not _env_path.exists(): if not _env_path.exists():
_env_path = Path("/etc/gotelegram-bot/.env") _env_path = Path("/opt/swiftgram/.env")
if _env_path.exists(): if _env_path.exists():
with open(_env_path, encoding="utf-8") as f: with open(_env_path, encoding="utf-8") as f:
for line in f: for line in f:
@@ -33,7 +35,7 @@ from telegram.ext import (
filters, filters,
) )
# ── Конфиг ──────────────────────────────────────────────────────────────────── # ── Конфигурация ─────────────────────────────────────────────────────────────
BOT_TOKEN = os.environ.get("BOT_TOKEN") BOT_TOKEN = os.environ.get("BOT_TOKEN")
_allowed = os.environ.get("ALLOWED_IDS", "").strip() _allowed = os.environ.get("ALLOWED_IDS", "").strip()
try: try:
@@ -41,8 +43,9 @@ try:
except ValueError: except ValueError:
ALLOWED_IDS = None ALLOWED_IDS = None
CONTAINER_NAME = "mtproto-proxy" CONTAINER_NAME = os.environ.get("CONTAINER_NAME", "swiftgram-proxy")
CONFIG_FILE = Path("/opt/gotelegram-bot/proxy.json") CONFIG_FILE = Path(os.environ.get("CONFIG_PATH", "/opt/swiftgram/proxy.json"))
DOMAINS = [ DOMAINS = [
"google.com", "wikipedia.org", "habr.com", "github.com", "google.com", "wikipedia.org", "habr.com", "github.com",
"coursera.org", "udemy.com", "medium.com", "stackoverflow.com", "coursera.org", "udemy.com", "medium.com", "stackoverflow.com",
@@ -50,19 +53,14 @@ DOMAINS = [
"lenta.ru", "rbc.ru", "ria.ru", "kommersant.ru", "lenta.ru", "rbc.ru", "ria.ru", "kommersant.ru",
"stepik.org", "duolingo.com", "khanacademy.org", "ted.com", "stepik.org", "duolingo.com", "khanacademy.org", "ted.com",
] ]
PROMO_LINK = "https://vk.cc/ct29NQ"
TIP_LINK = "https://pay.cloudtips.ru/p/7410814f"
# ── Утилиты ────────────────────────────────────────────────────────────────── # ── Утилиты ──────────────────────────────────────────────────────────────────
def _ok(uid: int) -> bool: def _ok(uid: int) -> bool:
return ALLOWED_IDS is None or uid in ALLOWED_IDS return ALLOWED_IDS is None or uid in ALLOWED_IDS
def _decode(data: bytes) -> str: def _decode(data: bytes) -> str:
return (data or b"").decode("utf-8", errors="replace").strip() return (data or b"").decode("utf-8", errors="replace").strip()
async def sh(*args: str, timeout: int = 60) -> tuple[int, str, str]: async def sh(*args: str, timeout: int = 60) -> tuple[int, str, str]:
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
*args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
@@ -75,560 +73,254 @@ async def sh(*args: str, timeout: int = 60) -> tuple[int, str, str]:
return -1, "", "Timeout" return -1, "", "Timeout"
return proc.returncode or 0, _decode(out), _decode(err) return proc.returncode or 0, _decode(out), _decode(err)
async def get_ip4() -> str:
async def get_ip() -> str:
for url in ("https://api.ipify.org", "https://icanhazip.com", "https://ifconfig.me"): for url in ("https://api.ipify.org", "https://icanhazip.com", "https://ifconfig.me"):
code, out, _ = await sh("curl", "-s", "-4", "--max-time", "5", url, timeout=8) code, out, _ = await sh("curl", "-s", "-4", "--max-time", "5", url, timeout=8)
if code == 0 and out: if code == 0 and out:
m = re.search(r"(\d{1,3}\.){3}\d{1,3}", out) m = re.search(r"(\d{1,3}\.){3}\d{1,3}", out)
if m: if m: return m.group(0)
return m.group(0)
return "0.0.0.0" return "0.0.0.0"
async def get_ip6() -> str:
code, out, _ = await sh("curl", "-s", "-6", "--max-time", "5", "https://api6.ipify.org", timeout=8)
if code == 0 and out:
return out.strip()
return ""
async def check_bbr() -> bool:
code, out, _ = await sh("sysctl", "net.ipv4.tcp_congestion_control", timeout=5)
return "bbr" in out.lower()
async def proxy_running() -> bool: async def proxy_running() -> bool:
code, out, _ = await sh("docker", "ps", "--format", "{{.Names}}", timeout=10) code, out, _ = await sh("docker", "ps", "--format", "{{.Names}}", timeout=10)
return code == 0 and CONTAINER_NAME in out return code == 0 and CONTAINER_NAME in out
async def docker_val(fmt: str) -> str: async def docker_val(fmt: str) -> str:
code, out, _ = await sh("docker", "inspect", CONTAINER_NAME, "--format", fmt, timeout=10) code, out, _ = await sh("docker", "inspect", CONTAINER_NAME, "--format", fmt, timeout=10)
return out.strip() if code == 0 else "" return out.strip() if code == 0 else ""
async def check_port(port: int) -> str | None: async def check_port(port: int) -> str | None:
"""Если порт занят — возвращает описание процесса; иначе None."""
# Пропускаем, если порт занят нашим же контейнером
if await proxy_running(): if await proxy_running():
hp = await docker_val("{{range $p,$c := .HostConfig.PortBindings}}{{(index $c 0).HostPort}} {{end}}") hp = await docker_val("{{range $p,$c := .HostConfig.PortBindings}}{{(index $c 0).HostPort}} {{end}}")
if str(port) in hp.split(): if str(port) in hp.split(): return None
return None
code, out, _ = await sh("ss", "-tlnp", timeout=5) code, out, _ = await sh("ss", "-tlnp", timeout=5)
if code != 0: if code != 0: code, out, _ = await sh("netstat", "-tlnp", timeout=5)
code, out, _ = await sh("netstat", "-tlnp", timeout=5)
for line in out.splitlines(): for line in out.splitlines():
if f":{port} " in line or f":{port}\t" in line: if f":{port} " in line or f":{port}\t" in line: return line
return line
return None return None
async def docker_containers_info() -> str:
code, out, _ = await sh("docker", "ps", "--format", "{{.Names}}\t{{.Image}}\t{{.Ports}}", timeout=10)
if code != 0 or not out:
return ""
return out
def save_config(data: dict) -> None: def save_config(data: dict) -> None:
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True) CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
CONFIG_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") CONFIG_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def load_config() -> dict: def load_config() -> dict:
if CONFIG_FILE.exists(): if CONFIG_FILE.exists():
try: try: return json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
return json.loads(CONFIG_FILE.read_text(encoding="utf-8")) except: pass
except Exception:
pass
return {} return {}
# ── Получение данных прокси ────────────────────────────────────────────────── # ── Получение данных прокси ──────────────────────────────────────────────────
async def proxy_info() -> dict | None: async def proxy_info() -> dict | None:
if not await proxy_running(): if not await proxy_running(): return None
return None
cmd_str = await docker_val("{{range .Config.Cmd}}{{.}} {{end}}") cmd_str = await docker_val("{{range .Config.Cmd}}{{.}} {{end}}")
secret = cmd_str.split()[-1] if cmd_str else "" secret = cmd_str.split()[-1] if cmd_str else ""
hp = await docker_val("{{range $p,$c := .HostConfig.PortBindings}}{{(index $c 0).HostPort}}{{end}}") hp = await docker_val("{{range $p,$c := .HostConfig.PortBindings}}{{(index $c 0).HostPort}}{{end}}")
port = hp or "443" port = hp or "443"
ip = await get_ip() ip4 = await get_ip4()
link = f"tg://proxy?server={ip}&port={port}&secret={secret}" ip6 = await get_ip6()
cfg = load_config() cfg = load_config()
return {"ip": ip, "port": port, "secret": secret, "link": link, "domain": cfg.get("domain", "—")} return {
"ip4": ip4, "ip6": ip6, "port": port, "secret": secret,
"domain": cfg.get("domain", "—"),
"link4": f"tg://proxy?server={ip4}&port={port}&secret={secret}",
"link6": f"tg://proxy?server={ip6}&port={port}&secret={secret}" if ip6 else None
}
# ── Меню ─────────────────────────────────────────────────────────────────────
# ── Главное меню (кнопки) ────────────────────────────────────────────────────
def main_menu_kb() -> InlineKeyboardMarkup: def main_menu_kb() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup([ return InlineKeyboardMarkup([
[InlineKeyboardButton("🔧 Установить / Обновить", callback_data="menu_install")], [InlineKeyboardButton("🔧 Установить / Обновить", callback_data="menu_install")],
[InlineKeyboardButton("📊 Статус", callback_data="menu_status"), [InlineKeyboardButton("📊 Статус", callback_data="menu_status"),
InlineKeyboardButton("🔗 Ссылка", callback_data="menu_link")], InlineKeyboardButton("🔗 Ссылка", callback_data="menu_link")],
[InlineKeyboardButton("📤 Поделиться ключом", callback_data="menu_share")], [InlineKeyboardButton("📤 Поделиться", callback_data="menu_share")],
[InlineKeyboardButton("🔄 Перезапуск", callback_data="menu_restart"), [InlineKeyboardButton("🔄 Рестарт", callback_data="menu_restart"),
InlineKeyboardButton("📋 Логи", callback_data="menu_logs")], InlineKeyboardButton("📋 Логи", callback_data="menu_logs")],
[InlineKeyboardButton("🗑 Удалить", callback_data="menu_remove"), [InlineKeyboardButton("🗑 Удалить прокси", callback_data="menu_remove")],
InlineKeyboardButton("🏷 Промо", callback_data="menu_promo")],
]) ])
HELP_TEXT = ( HELP_TEXT = (
"🚀 <b>GoTelegram MTProxy Bot</b>\n\n" "🚀 <b>SwiftGram MTProxy Manager</b>\n\n"
"Управление MTProxy (Fake TLS) на сервере.\n" "Управление прокси на сервере.\n"
"TCP + UDP (звонки) поддержаны.\n\n" "• TCP + UDP (звонки) активны\n"
"• IPv6 поддержка включена\n"
"• BBR оптимизация стека\n\n"
"Используйте кнопки ниже или команды:\n" "Используйте кнопки ниже или команды:\n"
"/install /status /link /share /restart /logs /remove /promo" "/install /status /link /share /restart /logs /remove"
) )
async def start(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: async def start(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
if not update.effective_user: if not update.effective_user or not _ok(update.effective_user.id):
return return
if not _ok(update.effective_user.id):
msg = update.message or (update.callback_query and update.callback_query.message) msg = update.message or (update.callback_query and update.callback_query.message)
if msg:
await msg.reply_text("⛔ Доступ запрещён.")
return
if update.message: if update.message:
await update.message.reply_text(HELP_TEXT, parse_mode="HTML", reply_markup=main_menu_kb()) await update.message.reply_text(HELP_TEXT, parse_mode="HTML", reply_markup=main_menu_kb())
elif update.callback_query: elif update.callback_query:
await update.callback_query.edit_message_text(HELP_TEXT, parse_mode="HTML", reply_markup=main_menu_kb()) await update.callback_query.edit_message_text(HELP_TEXT, parse_mode="HTML", reply_markup=main_menu_kb())
# ── Команды ──────────────────────────────────────────────────────────────────
# ── Статус ───────────────────────────────────────────────────────────────────
async def cmd_status(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: async def cmd_status(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
msg = update.message or (update.callback_query and update.callback_query.message) if not update.effective_user or not _ok(update.effective_user.id): return
if not update.effective_user or not msg:
return
if not _ok(update.effective_user.id):
await msg.reply_text("⛔")
return
info = await proxy_info() info = await proxy_info()
if not info: if not info:
text = "❌ Прокси не запущен.\nНажмите <b>Установить</b> для настройки." text = "❌ <b>Прокси не запущен.</b>\nИспользуйте команду /install"
else: else:
containers = await docker_containers_info() bbr = "✅ Активен" if await check_bbr() else "❌ Выключен"
other = "\n".join(l for l in containers.splitlines() if CONTAINER_NAME not in l)
text = ( text = (
"✅ <b>Прокси работает</b>\n\n" "✅ <b>SwiftGram работает</b>\n\n"
f"IP: <code>{html.escape(info['ip'])}</code>\n" f"🌐 IPv4: <code>{info['ip4']}</code>\n"
f"Порт: <code>{html.escape(info['port'])}</code>\n" f"🌐 IPv6: <code>{info['ip6'] or 'не найден'}</code>\n"
f"Домен: <code>{html.escape(info['domain'])}</code>\n" f"🔌 Порт: <code>{info['port']}</code>\n"
f"Secret: <code>{html.escape(info['secret'])}</code>\n\n" f"🎭 Домен: <code>{html.escape(info['domain'])}</code>\n"
f"Ссылка:\n<code>{html.escape(info['link'])}</code>" f"🚀 BBR: <code>{bbr}</code>\n"
f"📞 Звонки: <code>✅ UDP открыт</code>\n\n"
f"Secret: <code>{info['secret']}</code>"
) )
if other:
text += f"\n\n📦 <b>Другие контейнеры:</b>\n<pre>{html.escape(other)}</pre>"
kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]]) kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]])
if update.callback_query: if update.callback_query: await update.callback_query.edit_message_text(text, parse_mode="HTML", reply_markup=kb)
await update.callback_query.edit_message_text(text, parse_mode="HTML", reply_markup=kb) else: await update.message.reply_text(text, parse_mode="HTML", reply_markup=kb)
else:
await msg.reply_text(text, parse_mode="HTML", reply_markup=kb)
# ── Ссылка ───────────────────────────────────────────────────────────────────
async def cmd_link(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: async def cmd_link(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
msg = update.message or (update.callback_query and update.callback_query.message) if not update.effective_user or not _ok(update.effective_user.id): return
if not update.effective_user or not msg:
return
if not _ok(update.effective_user.id):
return
info = await proxy_info() info = await proxy_info()
if not info: if not info: text = "❌ Прокси не запущен."
text = "❌ Прокси не запущен."
else: else:
text = f"<code>{html.escape(info['link'])}</code>" text = f"<b>Ваша ссылка:</b>\n<code>{info['link4']}</code>"
if info['link6']: text += f"\n\n<b>IPv6 ссылка:</b>\n<code>{info['link6']}</code>"
kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]]) kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]])
if update.callback_query: if update.callback_query: await update.callback_query.edit_message_text(text, parse_mode="HTML", reply_markup=kb)
await update.callback_query.edit_message_text(text, parse_mode="HTML", reply_markup=kb) else: await update.message.reply_text(text, parse_mode="HTML", reply_markup=kb)
else:
await msg.reply_text(text, parse_mode="HTML", reply_markup=kb)
# ── Поделиться ключом ────────────────────────────────────────────────────────
async def cmd_share(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: async def cmd_share(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
msg = update.message or (update.callback_query and update.callback_query.message) if not update.effective_user or not _ok(update.effective_user.id): return
if not update.effective_user or not msg:
return
if not _ok(update.effective_user.id):
return
info = await proxy_info() info = await proxy_info()
if not info: if not info: return
text = "❌ Прокси не запущен."
kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]])
if update.callback_query:
await update.callback_query.edit_message_text(text, reply_markup=kb)
else:
await msg.reply_text(text, reply_markup=kb)
return
# tg://proxy ссылка, которую Telegram распознает при пересылке
tg_link = info["link"]
# Красивое сообщение для пересылки
share_text = ( share_text = (
f"🔐 <b>MTProxy для Telegram</b>\n\n" f"🔐 <b>MTProxy для Telegram</b>\n\n"
f"🌍 Сервер: <code>{html.escape(info['ip'])}</code>\n" f"🌍 Сервер: <code>{info['ip4']}</code>\n"
f"🔌 Порт: <code>{html.escape(info['port'])}</code>\n" f"🔌 Порт: <code>{info['port']}</code>\n"
f"🔑 Secret: <code>{html.escape(info['secret'])}</code>\n\n" f"🔑 Secret: <code>{info['secret']}</code>\n\n"
f"👉 <b>Подключиться одним нажатием:</b>\n" f"👉 <b>Подключиться:</b>\n{info['link4']}"
f"{html.escape(tg_link)}\n\n"
f"Просто нажмите на ссылку или перешлите это сообщение."
) )
kb = InlineKeyboardMarkup([ kb = InlineKeyboardMarkup([
[InlineKeyboardButton("📤 Переслать другу", switch_inline_query=tg_link)], [InlineKeyboardButton("📤 Переслать другу", switch_inline_query=info['link4'])],
[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")], [InlineKeyboardButton("◀️ Меню", callback_data="menu_main")],
]) ])
if update.callback_query: if update.callback_query: await update.callback_query.edit_message_text(share_text, parse_mode="HTML", reply_markup=kb)
await update.callback_query.edit_message_text(share_text, parse_mode="HTML", reply_markup=kb) else: await update.message.reply_text(share_text, parse_mode="HTML", reply_markup=kb)
else:
await msg.reply_text(share_text, parse_mode="HTML", reply_markup=kb)
# ── Удалить ──────────────────────────────────────────────────────────────────
async def cmd_remove(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
msg = update.message or (update.callback_query and update.callback_query.message)
if not update.effective_user or not msg:
return
if not _ok(update.effective_user.id):
return
chat = msg.chat
if update.callback_query:
await update.callback_query.edit_message_text("⏳ Удаляю прокси...")
else:
await chat.send_message("⏳ Удаляю прокси...")
await sh("docker", "stop", CONTAINER_NAME, timeout=15)
await sh("docker", "rm", CONTAINER_NAME, timeout=10)
text = "✅ Прокси удалён." if not await proxy_running() else "⚠️ Не удалось удалить."
kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]])
await chat.send_message(text, reply_markup=kb)
# ── Рестарт ──────────────────────────────────────────────────────────────────
async def cmd_restart(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: async def cmd_restart(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
msg = update.message or (update.callback_query and update.callback_query.message) if not update.effective_user or not _ok(update.effective_user.id): return
if not update.effective_user or not msg: if update.callback_query: await update.callback_query.answer("Перезапуск...")
return
if not _ok(update.effective_user.id):
return
if not await proxy_running():
kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]])
if update.callback_query:
await update.callback_query.edit_message_text("❌ Прокси не запущен.", reply_markup=kb)
else:
await msg.reply_text("❌ Прокси не запущен.", reply_markup=kb)
return
chat = msg.chat
if update.callback_query:
await update.callback_query.edit_message_text("⏳ Перезапуск...")
code, _, err = await sh("docker", "restart", CONTAINER_NAME, timeout=30) code, _, err = await sh("docker", "restart", CONTAINER_NAME, timeout=30)
text = "✅ Перезапущен." if code == 0 else f"❌ Ошибка: {err or 'unknown'}" text = "✅ Контейнер перезапущен." if code == 0 else f"❌ Ошибка: {err}"
kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]]) kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]])
await chat.send_message(text, reply_markup=kb) if update.callback_query: await update.callback_query.edit_message_text(text, reply_markup=kb)
else: await update.message.reply_text(text, reply_markup=kb)
# ── Логи ─────────────────────────────────────────────────────────────────────
async def cmd_logs(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: async def cmd_logs(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
msg = update.message or (update.callback_query and update.callback_query.message) if not update.effective_user or not _ok(update.effective_user.id): return
if not update.effective_user or not msg: code, out, err = await sh("docker", "logs", "--tail", "30", CONTAINER_NAME, timeout=15)
return text = f"<pre>{html.escape(out or err or 'Логов нет.')}</pre>"
if not _ok(update.effective_user.id):
return
if not await proxy_running():
kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]]) kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]])
if update.callback_query: if update.callback_query: await update.callback_query.edit_message_text(text, parse_mode="HTML", reply_markup=kb)
await update.callback_query.edit_message_text("❌ Прокси не запущен.", reply_markup=kb) else: await update.message.reply_text(text, parse_mode="HTML", reply_markup=kb)
else:
await msg.reply_text("❌ Прокси не запущен.", reply_markup=kb) async def cmd_remove(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
return if not update.effective_user or not _ok(update.effective_user.id): return
code, out, err = await sh("docker", "logs", "--tail", "40", CONTAINER_NAME, timeout=15) if update.callback_query: await update.callback_query.edit_message_text("⏳ Удаление контейнера...")
text = (out or "") + (("\n" + err) if err else "") or "Нет вывода." await sh("docker", "stop", CONTAINER_NAME)
if len(text) > 4000: await sh("docker", "rm", CONTAINER_NAME)
text = text[-4000:] text = "✅ Прокси успешно удален с сервера."
kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]]) kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]])
if update.callback_query: if update.callback_query: await update.callback_query.edit_message_text(text, reply_markup=kb)
await update.callback_query.edit_message_text(f"<pre>{html.escape(text)}</pre>", parse_mode="HTML", reply_markup=kb) else: await update.message.reply_text(text, reply_markup=kb)
else:
await msg.reply_text(f"<pre>{html.escape(text)}</pre>", parse_mode="HTML", reply_markup=kb)
# ── Промо ────────────────────────────────────────────────────────────────────
async def cmd_promo(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
msg = update.message or (update.callback_query and update.callback_query.message)
if not update.effective_user or not msg:
return
if not _ok(update.effective_user.id):
return
text = (
"💰 <b>Хостинг со скидкой до -60%</b>\n"
f"Ссылка: {PROMO_LINK}\n\n"
"Промокоды: OFF60, antenka20, antenka6, antenka12\n\n"
f"Донат: {TIP_LINK}"
)
kb = InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]])
if update.callback_query:
await update.callback_query.edit_message_text(text, parse_mode="HTML", reply_markup=kb)
else:
await msg.reply_text(text, parse_mode="HTML", reply_markup=kb)
# ── Установка: домен → порт → проверка → запуск ─────────────────────────────
# ── Установка ────────────────────────────────────────────────────────────────
async def install_step_domain(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: async def install_step_domain(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
msg = update.message or (update.callback_query and update.callback_query.message) if not update.effective_user or not _ok(update.effective_user.id): return
if not update.effective_user or not msg:
return
if not _ok(update.effective_user.id):
return
buttons = [] buttons = []
row = [] row = []
for i, d in enumerate(DOMAINS): for i, d in enumerate(DOMAINS[:10]): # Показываем первые 10 для компактности
row.append(InlineKeyboardButton(d, callback_data=f"dom_{i}")) row.append(InlineKeyboardButton(d, callback_data=f"dom_{i}"))
if len(row) == 2: if len(row) == 2: buttons.append(row); row = []
buttons.append(row) buttons.append([InlineKeyboardButton("◀️ Меню", callback_data="menu_main")])
row = [] text = "🌐 <b>Выберите домен маскировки (Fake TLS):</b>"
if row: if update.callback_query: await update.callback_query.edit_message_text(text, parse_mode="HTML", reply_markup=InlineKeyboardMarkup(buttons))
buttons.append(row) else: await update.message.reply_text(text, parse_mode="HTML", reply_markup=InlineKeyboardMarkup(buttons))
text = "🌐 <b>Выберите домен для маскировки (Fake TLS):</b>"
if update.callback_query:
await update.callback_query.edit_message_text(text, parse_mode="HTML", reply_markup=InlineKeyboardMarkup(buttons))
else:
await msg.reply_text(text, parse_mode="HTML", reply_markup=InlineKeyboardMarkup(buttons))
async def install_step_port(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
query = update.callback_query
domain = ctx.user_data.get("install_domain", "google.com")
# Проверяем порт 443
busy_443 = await check_port(443)
busy_8443 = await check_port(8443)
rows = []
label_443 = "443 (рекомендуется)" if not busy_443 else "443 ⚠️ занят"
label_8443 = "8443" if not busy_8443 else "8443 ⚠️ занят"
rows.append([
InlineKeyboardButton(label_443, callback_data="port_443"),
InlineKeyboardButton(label_8443, callback_data="port_8443"),
])
rows.append([InlineKeyboardButton("◀️ Меню", callback_data="menu_main")])
port_info = ""
if busy_443:
port_info += f"\n⚠️ Порт 443 занят:\n<pre>{html.escape(busy_443[:300])}</pre>\n"
if busy_8443:
port_info += f"\n⚠️ Порт 8443 занят:\n<pre>{html.escape(busy_8443[:300])}</pre>\n"
text = (
f"Домен: <b>{html.escape(domain)}</b>\n\n"
"🔌 <b>Выберите порт</b> или введите свой (1-65535):"
f"{port_info}"
)
ctx.user_data["install_wait_port"] = True
await query.edit_message_text(text, parse_mode="HTML", reply_markup=InlineKeyboardMarkup(rows))
async def install_port_chosen(update: Update, ctx: ContextTypes.DEFAULT_TYPE, port_str: str) -> None:
"""Порт выбран кнопкой или текстом — проверяем и ставим."""
port = int(port_str)
msg = None
chat = None
if update.callback_query:
msg = update.callback_query.message
elif update.message:
msg = update.message
if not msg:
return
chat = msg.chat
# Проверка занятости
busy = await check_port(port)
if busy:
kb = InlineKeyboardMarkup([
[InlineKeyboardButton(f"Всё равно использовать {port}", callback_data=f"force_{port}")],
[InlineKeyboardButton("Выбрать другой порт", callback_data="reselect_port")],
[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")],
])
text = (
f"⚠️ <b>Порт {port} занят!</b>\n\n"
f"<pre>{html.escape(busy[:500])}</pre>\n\n"
"Можно использовать всё равно (если это ваш процесс) или выбрать другой."
)
if update.callback_query:
await update.callback_query.edit_message_text(text, parse_mode="HTML", reply_markup=kb)
else:
await chat.send_message(text, parse_mode="HTML", reply_markup=kb)
ctx.user_data["install_port"] = port_str
return
ctx.user_data["install_port"] = port_str
ctx.user_data["install_wait_port"] = False
await do_install(update, ctx)
async def do_install(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: async def do_install(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
domain = ctx.user_data.get("install_domain") or "google.com" domain = ctx.user_data.get("install_domain", "google.com")
port = ctx.user_data.get("install_port") or "443" # Бот делегирует установку системному вызову или запускает через docker напрямую
# Для безопасности в этой модульной версии мы просто дергаем docker run
port = "443"
if await check_port(443): port = "8443"
msg = None msg = update.callback_query.message if update.callback_query else update.message
if update.callback_query: await msg.reply_text(f"⏳ Начинаю установку SwiftGram на порт {port}...")
msg = update.callback_query.message
await msg.edit_text("⏳ Генерация secret и запуск контейнера...", reply_markup=None)
elif update.message:
msg = update.message
await msg.reply_text("⏳ Генерация secret и запуск контейнера...")
if not msg:
return
chat = msg.chat
# Docker check # Генерация секрета
code, _, _ = await sh("docker", "info", timeout=10) _, s_out, _ = await sh("docker", "run", "--rm", "nineseconds/mtg:2", "generate-secret", "--hex", domain)
if code != 0: secret = s_out.strip().split()[-1] if s_out else ""
await chat.send_message(
"❌ Docker не установлен или не запущен.\n"
"Установите: <code>curl -fsSL https://get.docker.com | sh</code>",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")]]),
)
return
# generate secret
code, secret_out, err = await sh(
"docker", "run", "--rm", "nineseconds/mtg:2", "generate-secret", "--hex", domain,
timeout=60,
)
if code != 0:
await chat.send_message(f"❌ Генерация secret: {err or secret_out}")
return
secret = secret_out.strip().split()[-1] if secret_out.strip() else ""
if not secret: if not secret:
await chat.send_message("❌ Пустой secret.") await msg.reply_text("❌ Ошибка генерации секрета.")
return return
# Остановка старого await sh("docker", "stop", CONTAINER_NAME)
await sh("docker", "stop", CONTAINER_NAME, timeout=15) await sh("docker", "rm", CONTAINER_NAME)
await sh("docker", "rm", CONTAINER_NAME, timeout=10)
# Запуск с TCP + UDP (UDP нужен для звонков Telegram)
# --network host не используем, чтобы не мешать Amnezia/3x-ui
code, _, err = await sh( code, _, err = await sh(
"docker", "run", "-d", "docker", "run", "-d", "--name", CONTAINER_NAME, "--restart", "always",
"--name", CONTAINER_NAME, "-p", f"{port}:{port}/tcp", "-p", f"{port}:{port}/udp",
"--restart", "always", "nineseconds/mtg:2", "simple-run", "-n", "1.1.1.1", "-i", "prefer-ipv4",
"-p", f"{port}:{port}/tcp", f"0.0.0.0:{port}", secret
"-p", f"{port}:{port}/udp",
"nineseconds/mtg:2",
"simple-run",
"-n", "1.1.1.1",
"-t", "1.0.0.1", # tag DNS для TLS
"-i", "prefer-ipv4",
f"0.0.0.0:{port}", secret,
timeout=90,
) )
if code != 0:
await chat.send_message(f"❌ Запуск контейнера: {err}")
return
if code == 0:
save_config({"domain": domain, "port": port, "secret": secret}) save_config({"domain": domain, "port": port, "secret": secret})
await msg.reply_text(f"✅ <b>SwiftGram установлен!</b>\nПорт: {port}\nДомен: {domain}", parse_mode="HTML")
await cmd_status(update, ctx)
else:
await msg.reply_text(f"❌ Ошибка: {err}")
ip = await get_ip() # ── Обработчики ──────────────────────────────────────────────────────────────
link = f"tg://proxy?server={ip}&port={port}&secret={secret}"
text = (
"✅ <b>Прокси установлен!</b>\n\n"
f"🌍 IP: <code>{html.escape(ip)}</code>\n"
f"🔌 Порт: <code>{html.escape(port)}</code> (TCP + UDP)\n"
f"🎭 Домен: <code>{html.escape(domain)}</code>\n"
f"🔑 Secret: <code>{html.escape(secret)}</code>\n\n"
f"👉 Ссылка:\n<code>{html.escape(link)}</code>\n\n"
"📞 Звонки в Telegram поддержаны (UDP)."
)
kb = InlineKeyboardMarkup([
[InlineKeyboardButton("📤 Поделиться ключом", callback_data="menu_share")],
[InlineKeyboardButton("◀️ Меню", callback_data="menu_main")],
])
await chat.send_message(text, parse_mode="HTML", reply_markup=kb)
for k in ("install_domain", "install_port", "install_wait_port"):
ctx.user_data.pop(k, None)
# ── Callback router ──────────────────────────────────────────────────────────
async def callback_handler(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: async def callback_handler(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
query = update.callback_query query = update.callback_query
if not query or not update.effective_user:
return
await query.answer() await query.answer()
if not _ok(update.effective_user.id): data = query.data
await query.edit_message_text("⛔ Доступ запрещён.") if data == "menu_main": await start(update, ctx)
return elif data == "menu_install": await install_step_domain(update, ctx)
elif data == "menu_status": await cmd_status(update, ctx)
data = query.data or "" elif data == "menu_link": await cmd_link(update, ctx)
elif data == "menu_share": await cmd_share(update, ctx)
if data == "menu_main": elif data == "menu_restart": await cmd_restart(update, ctx)
await start(update, ctx) elif data == "menu_logs": await cmd_logs(update, ctx)
elif data == "menu_install": elif data == "menu_remove": await cmd_remove(update, ctx)
await install_step_domain(update, ctx)
elif data == "menu_status":
await cmd_status(update, ctx)
elif data == "menu_link":
await cmd_link(update, ctx)
elif data == "menu_share":
await cmd_share(update, ctx)
elif data == "menu_restart":
await cmd_restart(update, ctx)
elif data == "menu_logs":
await cmd_logs(update, ctx)
elif data == "menu_remove":
await cmd_remove(update, ctx)
elif data == "menu_promo":
await cmd_promo(update, ctx)
elif data.startswith("dom_"): elif data.startswith("dom_"):
try:
idx = int(data[4:]) idx = int(data[4:])
except ValueError:
await query.edit_message_text("❌ Ошибка. /install")
return
if not (0 <= idx < len(DOMAINS)):
await query.edit_message_text("❌ Неверный выбор. /install")
return
ctx.user_data["install_domain"] = DOMAINS[idx] ctx.user_data["install_domain"] = DOMAINS[idx]
await install_step_port(update, ctx)
elif data == "port_443":
await install_port_chosen(update, ctx, "443")
elif data == "port_8443":
await install_port_chosen(update, ctx, "8443")
elif data.startswith("force_"):
port_str = data[6:]
ctx.user_data["install_port"] = port_str
ctx.user_data["install_wait_port"] = False
await do_install(update, ctx) await do_install(update, ctx)
elif data == "reselect_port":
await install_step_port(update, ctx)
# ── Ввод порта текстом ──────────────────────────────────────────────────────
async def text_handler(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message or not ctx.user_data.get("install_wait_port"):
return
text = (update.message.text or "").strip()
if not re.match(r"^\d+$", text):
return
port = int(text)
if not (1 <= port <= 65535):
await update.message.reply_text("Введите число от 1 до 65535.")
return
await install_port_chosen(update, ctx, str(port))
# ── main ─────────────────────────────────────────────────────────────────────
def main() -> None: def main() -> None:
if not BOT_TOKEN: if not BOT_TOKEN: raise SystemExit("Задайте BOT_TOKEN в .env")
raise SystemExit("Задайте BOT_TOKEN в .env")
app = Application.builder().token(BOT_TOKEN).build() app = Application.builder().token(BOT_TOKEN).build()
app.add_handler(CommandHandler("start", start)) app.add_handler(CommandHandler("start", start))
app.add_handler(CommandHandler("help", start))
app.add_handler(CommandHandler("install", install_step_domain)) app.add_handler(CommandHandler("install", install_step_domain))
app.add_handler(CommandHandler("status", cmd_status)) app.add_handler(CommandHandler("status", cmd_status))
app.add_handler(CommandHandler("link", cmd_link))
app.add_handler(CommandHandler("share", cmd_share))
app.add_handler(CommandHandler("remove", cmd_remove))
app.add_handler(CommandHandler("restart", cmd_restart)) app.add_handler(CommandHandler("restart", cmd_restart))
app.add_handler(CommandHandler("logs", cmd_logs)) app.add_handler(CommandHandler("remove", cmd_remove))
app.add_handler(CommandHandler("promo", cmd_promo))
app.add_handler(CallbackQueryHandler(callback_handler)) app.add_handler(CallbackQueryHandler(callback_handler))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, text_handler)) app.run_polling()
app.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == "__main__": if __name__ == "__main__":
main() main()