diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index a44eb7d..76d7216 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -303,7 +303,7 @@ jobs: Maintainer: Flowseal Depends: libgtk-3-0, libayatana-appindicator3-1, python3-tk Description: Telegram Desktop WebSocket Bridge Proxy - SOCKS5/WebSocket bridge proxy for Telegram Desktop with tray UI. + MTProto/WebSocket bridge proxy for Telegram Desktop with tray UI. EOF dpkg-deb --build --root-owner-group \ diff --git a/Dockerfile b/Dockerfile index dae44d2..b0d9462 100644 --- a/Dockerfile +++ b/Dockerfile @@ -23,7 +23,7 @@ ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 \ PATH=/opt/venv/bin:$PATH \ TG_WS_PROXY_HOST=0.0.0.0 \ - TG_WS_PROXY_PORT=1080 \ + TG_WS_PROXY_PORT=1443 \ TG_WS_PROXY_DC_IPS="2:149.154.167.220 4:149.154.167.220" RUN apt-get update \ @@ -39,7 +39,7 @@ COPY README.md LICENSE ./ USER app -EXPOSE 1080/tcp +EXPOSE 1443/tcp ENTRYPOINT ["/usr/bin/tini", "--", "/bin/sh", "-lc", "set -eu; args=\"--host ${TG_WS_PROXY_HOST} --port ${TG_WS_PROXY_PORT}\"; for dc in ${TG_WS_PROXY_DC_IPS}; do args=\"$args --dc-ip $dc\"; done; exec python -u proxy/tg_ws_proxy.py $args \"$@\"", "--"] CMD [] diff --git a/README.md b/README.md index 6881dc8..c99becd 100644 --- a/README.md +++ b/README.md @@ -12,17 +12,17 @@ # TG WS Proxy -**Локальный SOCKS5-прокси** для Telegram Desktop, который **ускоряет работу Telegram**, перенаправляя трафик через WebSocket-соединения. Данные передаются в том же зашифрованном виде, а для работы не нужны сторонние сервера. +**Локальный MTProto-прокси** для Telegram Desktop, который **ускоряет работу Telegram**, перенаправляя трафик через WebSocket-соединения. Данные передаются в том же зашифрованном виде, а для работы не нужны сторонние сервера. image ## Как это работает ``` -Telegram Desktop → SOCKS5 (127.0.0.1:1080) → TG WS Proxy → WSS → Telegram DC +Telegram Desktop → MTProto Proxy (127.0.0.1:1443) → WebSocket → Telegram DC ``` -1. Приложение поднимает локальный SOCKS5-прокси на `127.0.0.1:1080` +1. Приложение поднимает MTProto прокси на `127.0.0.1:1443` 2. Перехватывает подключения к IP-адресам Telegram 3. Извлекает DC ID из MTProto obfuscation init-пакета 4. Устанавливает WebSocket (TLS) соединение к соответствующему DC через домены Telegram @@ -38,7 +38,7 @@ Telegram Desktop → SOCKS5 (127.0.0.1:1080) → TG WS Proxy → WSS → Telegra **Меню трея:** -- **Открыть в Telegram** — автоматически настроить прокси через `tg://socks` ссылку +- **Открыть в Telegram** — автоматически настроить прокси через `tg://proxy` ссылку - **Перезапустить прокси** — перезапуск без выхода из приложения - **Настройки...** — GUI-редактор конфигурации (в т.ч. версия приложения, опциональная проверка обновлений с GitHub) - **Открыть логи** — открыть файл логов @@ -86,7 +86,7 @@ chmod +x TgWsProxy_linux_amd64 ### Консольный proxy -Для запуска только SOCKS5/WebSocket proxy без tray-интерфейса достаточно базовой установки: +Для запуска только proxy без tray-интерфейса достаточно базовой установки: ```bash pip install -e . @@ -124,9 +124,15 @@ tg-ws-proxy [--port PORT] [--host HOST] [--dc-ip DC:IP ...] [-v] | Аргумент | По умолчанию | Описание | |---|---|---| -| `--port` | `1080` | Порт SOCKS5-прокси | -| `--host` | `127.0.0.1` | Хост SOCKS5-прокси | +| `--port` | `1443` | Порт прокси | +| `--host` | `127.0.0.1` | Хост прокси | +| `--secret` | `random` | 32 hex chars secret для авторизации клиентов | | `--dc-ip` | `2:149.154.167.220`, `4:149.154.167.220` | Целевой IP для DC (можно указать несколько раз) | +| `--buf-kb` | `256` | Размер буфера в КБ +| `--pool-size` | `4` | Количество заготовленных соединений на каждый DC +| `--log-file` | выкл. | Путь до файла, в который сохранять логи +| `--log-max-mb` | `5` | Максимальный размер файла логов в МБ (после идёт перезапись) +| `--log-backups` | `0` | Количество сохранений логов после перезаписи | `-v`, `--verbose` | выкл. | Подробное логирование (DEBUG) | **Примеры:** @@ -166,10 +172,10 @@ tg-ws-proxy-tray-linux = "linux:main" 1. Telegram → **Настройки** → **Продвинутые настройки** → **Тип подключения** → **Прокси** 2. Добавить прокси: - - **Тип:** SOCKS5 - - **Сервер:** `127.0.0.1` - - **Порт:** `1080` - - **Логин/Пароль:** оставить пустыми + - **Тип:** MTProto + - **Сервер:** `127.0.0.1` (или переопределенный вами) + - **Порт:** `1443` (или переопределенный вами) + - **Secret:** из настроек или логов ## Конфигурация @@ -182,7 +188,8 @@ Tray-приложение хранит данные в: ```json { "host": "127.0.0.1", - "port": 1080, + "port": 1443, + "secret": "...", "dc_ip": [ "2:149.154.167.220", "4:149.154.167.220" diff --git a/linux.py b/linux.py index fe59ad6..8311040 100644 --- a/linux.py +++ b/linux.py @@ -20,7 +20,9 @@ import pystray from PIL import Image, ImageDraw, ImageFont import proxy.tg_ws_proxy as tg_ws_proxy +from proxy.tg_ws_proxy import proxy_config from proxy import __version__ + from utils.default_config import default_tray_config from ui.ctk_tray_ui import ( install_tray_config_buttons, @@ -33,7 +35,7 @@ from ui.ctk_theme import ( CONFIG_DIALOG_FRAME_PAD, CONFIG_DIALOG_SIZE, FIRST_RUN_SIZE, - create_ctk_root, + create_ctk_toplevel, ctk_theme_for_platform, main_content_frame, ) @@ -56,6 +58,9 @@ _config: dict = {} _exiting: bool = False _lock_file_path: Optional[Path] = None +_ctk_root = None +_ctk_root_ready = threading.Event() + log = logging.getLogger("tg-ws-tray") @@ -246,18 +251,59 @@ def _apply_linux_ctk_window_icon(root) -> None: root.iconphoto(False, root._ctk_icon_photo) -def _run_proxy_thread( - port: int, dc_opt: Dict[int, str], verbose: bool, host: str = "127.0.0.1" -): +def _ensure_ctk_thread() -> bool: + """Start the persistent hidden CTk root in its own thread (once).""" + global _ctk_root + if _ctk_root_ready.is_set(): + return True + + def _run(): + global _ctk_root + from ui.ctk_theme import ( + apply_ctk_appearance, + _install_tkinter_variable_del_guard, + ) + _install_tkinter_variable_del_guard() + apply_ctk_appearance(ctk) + _ctk_root = ctk.CTk() + _ctk_root.withdraw() + _ctk_root_ready.set() + _ctk_root.mainloop() + + threading.Thread(target=_run, daemon=True, name="ctk-root").start() + _ctk_root_ready.wait(timeout=5.0) + return _ctk_root is not None + + +def _ctk_run_dialog(build_fn) -> None: + """Schedule build_fn(done_event) on the CTk thread and block until done_event is set.""" + if _ctk_root is None: + return + done = threading.Event() + + def _invoke(): + try: + build_fn(done) + except Exception: + log.exception("CTk dialog failed") + done.set() + + _ctk_root.after(0, _invoke) + done.wait() + + +def _run_proxy_thread(): global _async_stop + loop = _asyncio.new_event_loop() _asyncio.set_event_loop(loop) + stop_ev = _asyncio.Event() _async_stop = (loop, stop_ev) try: loop.run_until_complete( - tg_ws_proxy._run(port, dc_opt, stop_event=stop_ev, host=host) + tg_ws_proxy._run(stop_event=stop_ev) ) except Exception as exc: log.error("Proxy thread crashed: %s", exc) @@ -279,27 +325,29 @@ def start_proxy(): cfg = _config port = cfg.get("port", DEFAULT_CONFIG["port"]) host = cfg.get("host", DEFAULT_CONFIG["host"]) + secret = cfg.get("secret", DEFAULT_CONFIG["secret"]) dc_ip_list = cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"]) - verbose = cfg.get("verbose", False) + buf_kb = cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"]) + pool_size = cfg.get("pool_size", DEFAULT_CONFIG["pool_size"]) try: - dc_opt = tg_ws_proxy.parse_dc_ip_list(dc_ip_list) + dc_redirects = tg_ws_proxy.parse_dc_ip_list(dc_ip_list) except ValueError as e: log.error("Bad config dc_ip: %s", e) _show_error(f"Ошибка конфигурации:\n{e}") return - log.info("Starting proxy on %s:%d ...", host, port) + proxy_config.port = port + proxy_config.host = host + proxy_config.secret = secret + proxy_config.dc_redirects = dc_redirects + proxy_config.buffer_size = max(4, buf_kb) * 1024 + proxy_config.pool_size = max(0, pool_size) - buf_kb = cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"]) - pool_size = cfg.get("pool_size", DEFAULT_CONFIG["pool_size"]) - tg_ws_proxy._RECV_BUF = max(4, buf_kb) * 1024 - tg_ws_proxy._SEND_BUF = tg_ws_proxy._RECV_BUF - tg_ws_proxy._WS_POOL_SIZE = max(0, pool_size) + log.info("Starting proxy on %s:%d ...", host, port) _proxy_thread = threading.Thread( target=_run_proxy_thread, - args=(port, dc_opt, verbose, host), daemon=True, name="proxy", ) @@ -389,7 +437,9 @@ def _maybe_notify_update_async(): def _on_open_in_telegram(icon=None, item=None): host = _config.get("host", DEFAULT_CONFIG["host"]) port = _config.get("port", DEFAULT_CONFIG["port"]) - url = f"tg://socks?server={host}&port={port}" + secret = _config.get("secret", DEFAULT_CONFIG["secret"]) + + url = f"tg://proxy?server={host}&port={port}&secret=dd{secret}" log.info("Copying %s", url) try: @@ -412,75 +462,67 @@ def _on_edit_config(icon=None, item=None): def _edit_config_dialog(): - if ctk is None: + if not _ensure_ctk_thread(): _show_error("customtkinter не установлен.") return cfg = dict(_config) - theme = ctk_theme_for_platform() - w, h = CONFIG_DIALOG_SIZE + def _build(done: threading.Event): + theme = ctk_theme_for_platform() + w, h = CONFIG_DIALOG_SIZE + root = create_ctk_toplevel( + ctk, + title="TG WS Proxy — Настройки", + width=w, + height=h, + theme=theme, + after_create=_apply_linux_ctk_window_icon, + ) - root = create_ctk_root( - ctk, - title="TG WS Proxy — Настройки", - width=w, - height=h, - theme=theme, - after_create=_apply_linux_ctk_window_icon, - ) + fpx, fpy = CONFIG_DIALOG_FRAME_PAD + frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy) + scroll, footer = tray_settings_scroll_and_footer(ctk, frame, theme) + widgets = install_tray_config_form( + ctk, scroll, theme, cfg, DEFAULT_CONFIG, + show_autostart=False, + ) - fpx, fpy = CONFIG_DIALOG_FRAME_PAD - frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy) - - scroll, footer = tray_settings_scroll_and_footer(ctk, frame, theme) - - widgets = install_tray_config_form( - ctk, scroll, theme, cfg, DEFAULT_CONFIG, - show_autostart=False, - ) - - def on_save(): - merged = validate_config_form( - widgets, DEFAULT_CONFIG, include_autostart=False) - if isinstance(merged, str): - _show_error(merged) - return - - new_cfg = merged - save_config(new_cfg) - _config.update(new_cfg) - log.info("Config saved: %s", new_cfg) - - _tray_icon.menu = _build_menu() - - from tkinter import messagebox - - if messagebox.askyesno( - "Перезапустить?", - "Настройки сохранены.\n\nПерезапустить прокси сейчас?", - parent=root, - ): - root.destroy() - restart_proxy() - else: + def _finish(): root.destroy() + done.set() - def on_cancel(): - root.destroy() + def on_save(): + merged = validate_config_form( + widgets, DEFAULT_CONFIG, include_autostart=False) + if isinstance(merged, str): + _show_error(merged) + return - install_tray_config_buttons( - ctk, footer, theme, on_save=on_save, on_cancel=on_cancel) + new_cfg = merged + save_config(new_cfg) + _config.update(new_cfg) + log.info("Config saved: %s", new_cfg) + _tray_icon.menu = _build_menu() - try: - root.mainloop() - finally: - import tkinter as tk - try: - if root.winfo_exists(): - root.destroy() - except tk.TclError: - pass + from tkinter import messagebox + do_restart = messagebox.askyesno( + "Перезапустить?", + "Настройки сохранены.\n\nПерезапустить прокси сейчас?", + parent=root) + _finish() + if do_restart: + threading.Thread( + target=restart_proxy, daemon=True).start() + + def on_cancel(): + _finish() + + root.protocol("WM_DELETE_WINDOW", on_cancel) + install_tray_config_buttons( + ctk, footer, theme, on_save=on_save, on_cancel=on_cancel) + + _ctk_run_dialog(_build) def _on_open_logs(icon=None, item=None): @@ -511,6 +553,12 @@ def _on_exit(icon=None, item=None): _exiting = True log.info("User requested exit") + if _ctk_root is not None: + try: + _ctk_root.after(0, _ctk_root.quit) + except Exception: + pass + def _force_exit(): time.sleep(3) os._exit(0) @@ -525,44 +573,38 @@ def _show_first_run(): _ensure_dirs() if FIRST_RUN_MARKER.exists(): return - - host = _config.get("host", DEFAULT_CONFIG["host"]) - port = _config.get("port", DEFAULT_CONFIG["port"]) - - if ctk is None: + if not _ensure_ctk_thread(): FIRST_RUN_MARKER.touch() return - theme = ctk_theme_for_platform() - w, h = FIRST_RUN_SIZE + host = _config.get("host", DEFAULT_CONFIG["host"]) + port = _config.get("port", DEFAULT_CONFIG["port"]) + secret = _config.get("secret", DEFAULT_CONFIG["secret"]) - root = create_ctk_root( - ctk, - title="TG WS Proxy", - width=w, - height=h, - theme=theme, - after_create=_apply_linux_ctk_window_icon, - ) + def _build(done: threading.Event): + theme = ctk_theme_for_platform() + w, h = FIRST_RUN_SIZE + root = create_ctk_toplevel( + ctk, + title="TG WS Proxy", + width=w, + height=h, + theme=theme, + after_create=_apply_linux_ctk_window_icon, + ) - def on_done(open_tg: bool): - FIRST_RUN_MARKER.touch() - root.destroy() - if open_tg: - _on_open_in_telegram() + def on_done(open_tg: bool): + FIRST_RUN_MARKER.touch() + root.destroy() + done.set() + if open_tg: + _on_open_in_telegram() - populate_first_run_window( - ctk, root, theme, host=host, port=port, on_done=on_done) + populate_first_run_window( + ctk, root, theme, host=host, port=port, secret=secret, + on_done=on_done) - try: - root.mainloop() - finally: - import tkinter as tk - try: - if root.winfo_exists(): - root.destroy() - except tk.TclError: - pass + _ctk_run_dialog(_build) def _has_ipv6_enabled() -> bool: @@ -617,9 +659,11 @@ def _build_menu(): return None host = _config.get("host", DEFAULT_CONFIG["host"]) port = _config.get("port", DEFAULT_CONFIG["port"]) + link_host = tg_ws_proxy.get_link_host(host) + return pystray.Menu( pystray.MenuItem( - f"Открыть в Telegram ({host}:{port})", _on_open_in_telegram, default=True + f"Открыть в Telegram ({link_host}:{port})", _on_open_in_telegram, default=True ), pystray.Menu.SEPARATOR, pystray.MenuItem("Перезапустить прокси", _on_restart), diff --git a/macos.py b/macos.py index b8660bf..8141e15 100644 --- a/macos.py +++ b/macos.py @@ -30,7 +30,9 @@ except ImportError: pyperclip = None import proxy.tg_ws_proxy as tg_ws_proxy +from proxy.tg_ws_proxy import proxy_config from proxy import __version__ + from utils.default_config import default_tray_config APP_NAME = "TgWsProxy" @@ -271,17 +273,18 @@ def _ask_yes_no_close(text: str, # Proxy lifecycle -def _run_proxy_thread(port: int, dc_opt: Dict[int, str], verbose: bool, - host: str = '127.0.0.1'): +def _run_proxy_thread(): global _async_stop + loop = _asyncio.new_event_loop() _asyncio.set_event_loop(loop) + stop_ev = _asyncio.Event() _async_stop = (loop, stop_ev) try: loop.run_until_complete( - tg_ws_proxy._run(port, dc_opt, stop_event=stop_ev, host=host)) + tg_ws_proxy._run(stop_event=stop_ev)) except Exception as exc: log.error("Proxy thread crashed: %s", exc) if "Address already in use" in str(exc): @@ -304,27 +307,29 @@ def start_proxy(): cfg = _config port = cfg.get("port", DEFAULT_CONFIG["port"]) host = cfg.get("host", DEFAULT_CONFIG["host"]) + secret = cfg.get("secret", DEFAULT_CONFIG["secret"]) dc_ip_list = cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"]) - verbose = cfg.get("verbose", False) + buf_kb = cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"]) + pool_size = cfg.get("pool_size", DEFAULT_CONFIG["pool_size"]) try: - dc_opt = tg_ws_proxy.parse_dc_ip_list(dc_ip_list) + dc_redirects = tg_ws_proxy.parse_dc_ip_list(dc_ip_list) except ValueError as e: log.error("Bad config dc_ip: %s", e) _show_error(f"Ошибка конфигурации:\n{e}") return - log.info("Starting proxy on %s:%d ...", host, port) + proxy_config.port = port + proxy_config.host = host + proxy_config.secret = secret + proxy_config.dc_redirects = dc_redirects + proxy_config.buffer_size = max(4, buf_kb) * 1024 + proxy_config.pool_size = max(0, pool_size) - buf_kb = cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"]) - pool_size = cfg.get("pool_size", DEFAULT_CONFIG["pool_size"]) - tg_ws_proxy._RECV_BUF = max(4, buf_kb) * 1024 - tg_ws_proxy._SEND_BUF = tg_ws_proxy._RECV_BUF - tg_ws_proxy._WS_POOL_SIZE = max(0, pool_size) + log.info("Starting proxy on %s:%d ...", host, port) _proxy_thread = threading.Thread( target=_run_proxy_thread, - args=(port, dc_opt, verbose, host), daemon=True, name="proxy") _proxy_thread.start() @@ -352,7 +357,9 @@ def restart_proxy(): def _on_open_in_telegram(_=None): host = _config.get("host", DEFAULT_CONFIG["host"]) port = _config.get("port", DEFAULT_CONFIG["port"]) - url = f"tg://socks?server={host}&port={port}" + secret = _config.get("secret", DEFAULT_CONFIG["secret"]) + + url = f"tg://proxy?server={host}&port={port}&secret=dd{secret}" log.info("Opening %s", url) try: result = subprocess.call(['open', url]) @@ -502,6 +509,17 @@ def _edit_config_dialog(): _show_error("Порт должен быть числом 1-65535") return + # Secret + secret_str = _osascript_input( + "MTProto Secret (32 hex символа):", + cfg.get("secret", DEFAULT_CONFIG["secret"])) + if secret_str is None: + return + secret_str = secret_str.strip().lower() + if len(secret_str) != 32 or not all(c in "0123456789abcdef" for c in secret_str): + _show_error("Secret должен быть строкой из 32 шестнадцатеричных символов.") + return + # DC-IP mappings dc_default = ", ".join(cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"])) dc_str = _osascript_input( @@ -548,6 +566,7 @@ def _edit_config_dialog(): new_cfg = { "host": host, "port": port, + "secret": secret_str, "dc_ip": dc_lines, "verbose": verbose, "buf_kb": adv.get("buf_kb", cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"])), @@ -576,7 +595,9 @@ def _show_first_run(): host = _config.get("host", DEFAULT_CONFIG["host"]) port = _config.get("port", DEFAULT_CONFIG["port"]) - tg_url = f"tg://socks?server={host}&port={port}" + secret = _config.get("secret", DEFAULT_CONFIG["secret"]) + + tg_url = f"tg://proxy?server={host}&port={port}&secret=dd{secret}" text = ( f"Прокси запущен и работает в строке меню.\n\n" @@ -586,7 +607,8 @@ def _show_first_run(): f" Или ссылка: {tg_url}\n\n" f"Вручную:\n" f" Настройки → Продвинутые → Тип подключения → Прокси\n" - f" SOCKS5 → {host} : {port} (без логина/пароля)\n\n" + f" MTProto → {host} : {port} \n" + f" Secret: dd{secret} \n\n" f"Открыть прокси в Telegram сейчас?" ) @@ -646,9 +668,10 @@ class TgWsProxyApp(_TgWsProxyAppBase): host = _config.get("host", DEFAULT_CONFIG["host"]) port = _config.get("port", DEFAULT_CONFIG["port"]) + link_host = tg_ws_proxy.get_link_host(host) self._open_tg_item = rumps.MenuItem( - f"Открыть в Telegram ({host}:{port})", + f"Открыть в Telegram ({link_host}:{port})", callback=_on_open_in_telegram) self._restart_item = rumps.MenuItem( "Перезапустить прокси", @@ -690,8 +713,10 @@ class TgWsProxyApp(_TgWsProxyAppBase): def update_menu_title(self): host = _config.get("host", DEFAULT_CONFIG["host"]) port = _config.get("port", DEFAULT_CONFIG["port"]) + link_host = tg_ws_proxy.get_link_host(host) + self._open_tg_item.title = ( - f"Открыть в Telegram ({host}:{port})") + f"Открыть в Telegram ({link_host}:{port})") def run_menubar(): diff --git a/proxy/tg_ws_proxy.py b/proxy/tg_ws_proxy.py new file mode 100644 index 0000000..0346699 --- /dev/null +++ b/proxy/tg_ws_proxy.py @@ -0,0 +1,1203 @@ +from __future__ import annotations + +import os +import ssl +import sys +import time +import base64 +import struct +import asyncio +import hashlib +import argparse +import logging +import logging.handlers +import socket as _socket + +from collections import deque +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Set, Tuple + +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + + +@dataclass +class ProxyConfig: + port: int = 1443 + host: str = '127.0.0.1' + secret: str = field(default_factory=lambda: os.urandom(16).hex()) + dc_redirects: Dict[int, str] = field(default_factory=lambda: {2: '149.154.167.220', 4: '149.154.167.220'}) + dc_overrides: Dict[int, int] = field(default_factory=lambda: {203: 2}) + buffer_size: int = 256 * 1024 + pool_size: int = 4 + + +proxy_config = ProxyConfig() +log = logging.getLogger('tg-mtproto-proxy') + +DC_DEFAULT_IPS: Dict[int, str] = { + 1: '149.154.175.50', + 2: '149.154.167.51', + 3: '149.154.175.100', + 4: '149.154.167.91', + 5: '149.154.171.5', + 203: '91.105.192.100' +} + +HANDSHAKE_LEN = 64 +SKIP_LEN = 8 +PREKEY_LEN = 32 +KEY_LEN = 32 +IV_LEN = 16 +PROTO_TAG_POS = 56 +DC_IDX_POS = 60 + +PROTO_TAG_ABRIDGED = b'\xef\xef\xef\xef' +PROTO_TAG_INTERMEDIATE = b'\xee\xee\xee\xee' +PROTO_TAG_SECURE = b'\xdd\xdd\xdd\xdd' + +PROTO_ABRIDGED_INT = 0xEFEFEFEF +PROTO_INTERMEDIATE_INT = 0xEEEEEEEE +PROTO_PADDED_INTERMEDIATE_INT = 0xDDDDDDDD + +RESERVED_FIRST_BYTES = {0xEF} +RESERVED_STARTS = {b'\x48\x45\x41\x44', b'\x50\x4F\x53\x54', + b'\x47\x45\x54\x20', b'\xee\xee\xee\xee', + b'\xdd\xdd\xdd\xdd', b'\x16\x03\x01\x02'} +RESERVED_CONTINUE = b'\x00\x00\x00\x00' + +DC_FAIL_COOLDOWN = 30.0 +WS_FAIL_TIMEOUT = 2.0 +ws_blacklist: Set[Tuple[int, bool]] = set() +dc_fail_until: Dict[Tuple[int, bool], float] = {} + +_st_BB = struct.Struct('>BB') +_st_BBH = struct.Struct('>BBH') +_st_BBQ = struct.Struct('>BBQ') +_st_BB4s = struct.Struct('>BB4s') +_st_BBH4s = struct.Struct('>BBH4s') +_st_BBQ4s = struct.Struct('>BBQ4s') +_st_H = struct.Struct('>H') +_st_Q = struct.Struct('>Q') +_st_I_le = struct.Struct(' bytes: + if not data: + return data + n = len(data) + mask_rep = (mask * (n // 4 + 1))[:n] + return (int.from_bytes(data, 'big') ^ + int.from_bytes(mask_rep, 'big')).to_bytes(n, 'big') + + +def get_link_host(host: str) -> Optional[str]: + if host == '0.0.0.0': + try: + with _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM) as _s: + _s.connect(('8.8.8.8', 80)) + link_host = _s.getsockname()[0] + except OSError: + link_host = '127.0.0.1' + return link_host + else: + return host + + +class WsHandshakeError(Exception): + def __init__(self, status_code: int, status_line: str, + headers: dict = None, location: str = None): + self.status_code = status_code + self.status_line = status_line + self.headers = headers or {} + self.location = location + super().__init__(f"HTTP {status_code}: {status_line}") + + @property + def is_redirect(self) -> bool: + return self.status_code in (301, 302, 303, 307, 308) + + +class RawWebSocket: + __slots__ = ('reader', 'writer', '_closed') + + OP_BINARY = 0x2 + OP_CLOSE = 0x8 + OP_PING = 0x9 + OP_PONG = 0xA + + def __init__(self, reader: asyncio.StreamReader, + writer: asyncio.StreamWriter): + self.reader = reader + self.writer = writer + self._closed = False + + @staticmethod + async def connect(ip: str, domain: str, path: str = '/apiws', + timeout: float = 10.0) -> 'RawWebSocket': + reader, writer = await asyncio.wait_for( + asyncio.open_connection(ip, 443, ssl=_ssl_ctx, + server_hostname=domain), + timeout=min(timeout, 10)) + _set_sock_opts(writer.transport) + + ws_key = base64.b64encode(os.urandom(16)).decode() + req = ( + f'GET {path} HTTP/1.1\r\n' + f'Host: {domain}\r\n' + f'Upgrade: websocket\r\n' + f'Connection: Upgrade\r\n' + f'Sec-WebSocket-Key: {ws_key}\r\n' + f'Sec-WebSocket-Version: 13\r\n' + f'Sec-WebSocket-Protocol: binary\r\n' + f'Origin: https://web.telegram.org\r\n' + f'User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' + f'AppleWebKit/537.36 (KHTML, like Gecko) ' + f'Chrome/131.0.0.0 Safari/537.36\r\n' + f'\r\n' + ) + writer.write(req.encode()) + await writer.drain() + + response_lines: list[str] = [] + try: + while True: + line = await asyncio.wait_for(reader.readline(), + timeout=timeout) + if line in (b'\r\n', b'\n', b''): + break + response_lines.append( + line.decode('utf-8', errors='replace').strip()) + except asyncio.TimeoutError: + writer.close() + raise + + if not response_lines: + writer.close() + raise WsHandshakeError(0, 'empty response') + + first_line = response_lines[0] + parts = first_line.split(' ', 2) + try: + status_code = int(parts[1]) if len(parts) >= 2 else 0 + except ValueError: + status_code = 0 + + if status_code == 101: + return RawWebSocket(reader, writer) + + headers: dict[str, str] = {} + for hl in response_lines[1:]: + if ':' in hl: + k, v = hl.split(':', 1) + headers[k.strip().lower()] = v.strip() + + writer.close() + raise WsHandshakeError(status_code, first_line, headers, + location=headers.get('location')) + + async def send(self, data: bytes): + if self._closed: + raise ConnectionError("WebSocket closed") + frame = self._build_frame(self.OP_BINARY, data, mask=True) + self.writer.write(frame) + await self.writer.drain() + + async def send_batch(self, parts: List[bytes]): + if self._closed: + raise ConnectionError("WebSocket closed") + for part in parts: + self.writer.write( + self._build_frame(self.OP_BINARY, part, mask=True)) + await self.writer.drain() + + async def recv(self) -> Optional[bytes]: + while not self._closed: + opcode, payload = await self._read_frame() + + if opcode == self.OP_CLOSE: + self._closed = True + try: + self.writer.write(self._build_frame( + self.OP_CLOSE, + payload[:2] if payload else b'', mask=True)) + await self.writer.drain() + except Exception: + pass + return None + + if opcode == self.OP_PING: + try: + self.writer.write( + self._build_frame(self.OP_PONG, payload, mask=True)) + await self.writer.drain() + except Exception: + pass + continue + + if opcode == self.OP_PONG: + continue + + if opcode in (0x1, 0x2): + return payload + continue + return None + + async def close(self): + if self._closed: + return + self._closed = True + try: + self.writer.write( + self._build_frame(self.OP_CLOSE, b'', mask=True)) + await self.writer.drain() + except Exception: + pass + try: + self.writer.close() + await self.writer.wait_closed() + except Exception: + pass + + @staticmethod + def _build_frame(opcode: int, data: bytes, + mask: bool = False) -> bytes: + length = len(data) + fb = 0x80 | opcode + if not mask: + if length < 126: + return _st_BB.pack(fb, length) + data + if length < 65536: + return _st_BBH.pack(fb, 126, length) + data + return _st_BBQ.pack(fb, 127, length) + data + mask_key = os.urandom(4) + masked = _xor_mask(data, mask_key) + if length < 126: + return _st_BB4s.pack(fb, 0x80 | length, mask_key) + masked + if length < 65536: + return _st_BBH4s.pack(fb, 0x80 | 126, length, mask_key) + masked + return _st_BBQ4s.pack(fb, 0x80 | 127, length, mask_key) + masked + + async def _read_frame(self) -> Tuple[int, bytes]: + hdr = await self.reader.readexactly(2) + opcode = hdr[0] & 0x0F + length = hdr[1] & 0x7F + if length == 126: + length = _st_H.unpack(await self.reader.readexactly(2))[0] + elif length == 127: + length = _st_Q.unpack(await self.reader.readexactly(8))[0] + if hdr[1] & 0x80: + mask_key = await self.reader.readexactly(4) + payload = await self.reader.readexactly(length) + return opcode, _xor_mask(payload, mask_key) + payload = await self.reader.readexactly(length) + return opcode, payload + + +def _human_bytes(n: int) -> str: + for unit in ('B', 'KB', 'MB', 'GB'): + if abs(n) < 1024: + return f"{n:.1f}{unit}" + n /= 1024 + return f"{n:.1f}TB" + + +def _try_handshake(handshake: bytes, secret: bytes) -> Optional[Tuple[int, bool, bytes, bytes]]: + dec_prekey_and_iv = handshake[SKIP_LEN:SKIP_LEN + PREKEY_LEN + IV_LEN] + dec_prekey = dec_prekey_and_iv[:PREKEY_LEN] + dec_iv = dec_prekey_and_iv[PREKEY_LEN:] + + dec_key = hashlib.sha256(dec_prekey + secret).digest() + + dec_iv_int = int.from_bytes(dec_iv, 'big') + decryptor = Cipher( + algorithms.AES(dec_key), modes.CTR(dec_iv_int.to_bytes(16, 'big')) + ).encryptor() + decrypted = decryptor.update(handshake) + + proto_tag = decrypted[PROTO_TAG_POS:PROTO_TAG_POS + 4] + if proto_tag not in (PROTO_TAG_ABRIDGED, PROTO_TAG_INTERMEDIATE, + PROTO_TAG_SECURE): + return None + + dc_idx = int.from_bytes( + decrypted[DC_IDX_POS:DC_IDX_POS + 2], 'little', signed=True) + + dc_id = abs(dc_idx) + is_media = dc_idx < 0 + + return dc_id, is_media, proto_tag, dec_prekey_and_iv + + +def _generate_relay_init(proto_tag: bytes, dc_idx: int) -> bytes: + while True: + rnd = bytearray(os.urandom(HANDSHAKE_LEN)) + if rnd[0] in RESERVED_FIRST_BYTES: + continue + if bytes(rnd[:4]) in RESERVED_STARTS: + continue + if rnd[4:8] == RESERVED_CONTINUE: + continue + break + + rnd_bytes = bytes(rnd) + + enc_key = rnd_bytes[SKIP_LEN:SKIP_LEN + PREKEY_LEN] + enc_iv = rnd_bytes[SKIP_LEN + PREKEY_LEN:SKIP_LEN + PREKEY_LEN + IV_LEN] + + encryptor = Cipher( + algorithms.AES(enc_key), modes.CTR(enc_iv) + ).encryptor() + + dc_bytes = struct.pack(' List[bytes]: + if not chunk: + return [] + if self._disabled: + return [chunk] + + self._cipher_buf.extend(chunk) + self._plain_buf.extend(self._dec.update(chunk)) + + parts = [] + while self._cipher_buf: + packet_len = self._next_packet_len() + if packet_len is None: + break + if packet_len <= 0: + parts.append(bytes(self._cipher_buf)) + self._cipher_buf.clear() + self._plain_buf.clear() + self._disabled = True + break + parts.append(bytes(self._cipher_buf[:packet_len])) + del self._cipher_buf[:packet_len] + del self._plain_buf[:packet_len] + return parts + + def flush(self) -> List[bytes]: + if not self._cipher_buf: + return [] + tail = bytes(self._cipher_buf) + self._cipher_buf.clear() + self._plain_buf.clear() + return [tail] + + def _next_packet_len(self) -> Optional[int]: + if not self._plain_buf: + return None + if self._proto == PROTO_ABRIDGED_INT: + return self._next_abridged_len() + if self._proto in (PROTO_INTERMEDIATE_INT, + PROTO_PADDED_INTERMEDIATE_INT): + return self._next_intermediate_len() + return 0 + + def _next_abridged_len(self) -> Optional[int]: + first = self._plain_buf[0] + if first in (0x7F, 0xFF): + if len(self._plain_buf) < 4: + return None + payload_len = int.from_bytes(self._plain_buf[1:4], 'little') * 4 + header_len = 4 + else: + payload_len = (first & 0x7F) * 4 + header_len = 1 + if payload_len <= 0: + return 0 + packet_len = header_len + payload_len + if len(self._plain_buf) < packet_len: + return None + return packet_len + + def _next_intermediate_len(self) -> Optional[int]: + if len(self._plain_buf) < 4: + return None + payload_len = _st_I_le.unpack_from(self._plain_buf, 0)[0] & 0x7FFFFFFF + if payload_len <= 0: + return 0 + packet_len = 4 + payload_len + if len(self._plain_buf) < packet_len: + return None + return packet_len + + +def _ws_domains(dc: int, is_media) -> List[str]: + dc = proxy_config.dc_overrides.get(dc, dc) + if is_media is None or is_media: + return [f'kws{dc}-1.web.telegram.org', f'kws{dc}.web.telegram.org'] + return [f'kws{dc}.web.telegram.org', f'kws{dc}-1.web.telegram.org'] + + +class Stats: + def __init__(self): + self.connections_total = 0 + self.connections_ws = 0 + self.connections_tcp_fallback = 0 + self.connections_bad = 0 + self.ws_errors = 0 + self.bytes_up = 0 + self.bytes_down = 0 + self.pool_hits = 0 + self.pool_misses = 0 + + def summary(self) -> str: + pool_total = self.pool_hits + self.pool_misses + pool_s = (f"{self.pool_hits}/{pool_total}" + if pool_total else "n/a") + return (f"total={self.connections_total} ws={self.connections_ws} " + f"tcp_fb={self.connections_tcp_fallback} " + f"bad={self.connections_bad} " + f"err={self.ws_errors} " + f"pool={pool_s} " + f"up={_human_bytes(self.bytes_up)} " + f"down={_human_bytes(self.bytes_down)}") + +_stats = Stats() + + +class _WsPool: + WS_POOL_MAX_AGE = 120.0 + + def __init__(self): + self._idle: Dict[Tuple[int, bool], deque] = {} + self._refilling: Set[Tuple[int, bool]] = set() + + async def get(self, dc: int, is_media: bool, + target_ip: str, domains: List[str] + ) -> Optional[RawWebSocket]: + key = (dc, is_media) + now = time.monotonic() + + bucket = self._idle.get(key) + if bucket is None: + bucket = deque() + self._idle[key] = bucket + while bucket: + ws, created = bucket.popleft() + age = now - created + if age > self.WS_POOL_MAX_AGE or ws._closed: + asyncio.create_task(self._quiet_close(ws)) + continue + _stats.pool_hits += 1 + log.debug("WS pool hit DC%d%s (age=%.1fs, left=%d)", + dc, 'm' if is_media else '', age, len(bucket)) + self._schedule_refill(key, target_ip, domains) + return ws + + _stats.pool_misses += 1 + self._schedule_refill(key, target_ip, domains) + return None + + def _schedule_refill(self, key, target_ip, domains): + if key in self._refilling: + return + self._refilling.add(key) + asyncio.create_task(self._refill(key, target_ip, domains)) + + async def _refill(self, key, target_ip, domains): + dc, is_media = key + try: + bucket = self._idle.setdefault(key, deque()) + needed = proxy_config.pool_size - len(bucket) + if needed <= 0: + return + tasks = [asyncio.create_task( + self._connect_one(target_ip, domains)) + for _ in range(needed)] + for t in tasks: + try: + ws = await t + if ws: + bucket.append((ws, time.monotonic())) + except Exception: + pass + log.debug("WS pool refilled DC%d%s: %d ready", + dc, 'm' if is_media else '', len(bucket)) + finally: + self._refilling.discard(key) + + @staticmethod + async def _connect_one(target_ip, domains) -> Optional[RawWebSocket]: + for domain in domains: + try: + return await RawWebSocket.connect( + target_ip, domain, timeout=8) + except WsHandshakeError as exc: + if exc.is_redirect: + continue + return None + except Exception: + return None + return None + + @staticmethod + async def _quiet_close(ws): + try: + await ws.close() + except Exception: + pass + + async def warmup(self, dc_redirects: Dict[int, Optional[str]]): + for dc, target_ip in dc_redirects.items(): + if target_ip is None: + continue + for is_media in (False, True): + domains = _ws_domains(dc, is_media) + self._schedule_refill((dc, is_media), target_ip, domains) + log.info("WS pool warmup started for %d DC(s)", len(dc_redirects)) + +_ws_pool = _WsPool() + + +async def _bridge_ws_reencrypt(reader, writer, ws: RawWebSocket, label, + dc=None, is_media=False, + clt_decryptor=None, clt_encryptor=None, + tg_encryptor=None, tg_decryptor=None, + splitter: _MsgSplitter = None): + """ + Bidirectional TCP(client) <-> WS(telegram) with re-encryption. + client ciphertext → decrypt(clt_key) → encrypt(tg_key) → WS + WS data → decrypt(tg_key) → encrypt(clt_key) → client TCP + """ + dc_tag = f"DC{dc}{'m' if is_media else ''}" if dc else "DC?" + + up_bytes = 0 + down_bytes = 0 + up_packets = 0 + down_packets = 0 + start_time = asyncio.get_event_loop().time() + + async def tcp_to_ws(): + nonlocal up_bytes, up_packets + try: + while True: + chunk = await reader.read(65536) + if not chunk: + if splitter: + tail = splitter.flush() + if tail: + await ws.send(tail[0]) + break + n = len(chunk) + _stats.bytes_up += n + up_bytes += n + up_packets += 1 + plain = clt_decryptor.update(chunk) + chunk = tg_encryptor.update(plain) + if splitter: + parts = splitter.split(chunk) + if not parts: + continue + if len(parts) > 1: + await ws.send_batch(parts) + else: + await ws.send(parts[0]) + else: + await ws.send(chunk) + except (asyncio.CancelledError, ConnectionError, OSError): + return + except Exception as e: + log.debug("[%s] tcp->ws ended: %s", label, e) + + async def ws_to_tcp(): + nonlocal down_bytes, down_packets + try: + while True: + data = await ws.recv() + if data is None: + break + n = len(data) + _stats.bytes_down += n + down_bytes += n + down_packets += 1 + plain = tg_decryptor.update(data) + data = clt_encryptor.update(plain) + writer.write(data) + await writer.drain() + except (asyncio.CancelledError, ConnectionError, OSError): + return + except Exception as e: + log.debug("[%s] ws->tcp ended: %s", label, e) + + tasks = [asyncio.create_task(tcp_to_ws()), + asyncio.create_task(ws_to_tcp())] + try: + await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + finally: + for t in tasks: + t.cancel() + for t in tasks: + try: + await t + except BaseException: + pass + elapsed = asyncio.get_event_loop().time() - start_time + log.info("[%s] %s WS session closed: " + "^%s (%d pkts) v%s (%d pkts) in %.1fs", + label, dc_tag, + _human_bytes(up_bytes), up_packets, + _human_bytes(down_bytes), down_packets, + elapsed) + try: + await ws.close() + except BaseException: + pass + try: + writer.close() + await writer.wait_closed() + except BaseException: + pass + + +async def _bridge_tcp_reencrypt(reader, writer, remote_reader, remote_writer, + label, dc=None, is_media=False, + clt_decryptor=None, clt_encryptor=None, + tg_encryptor=None, tg_decryptor=None): + """Bidirectional TCP <-> TCP with re-encryption.""" + + async def forward(src, dst_w, is_up): + try: + while True: + data = await src.read(65536) + if not data: + break + n = len(data) + if is_up: + _stats.bytes_up += n + plain = clt_decryptor.update(data) + data = tg_encryptor.update(plain) + else: + _stats.bytes_down += n + plain = tg_decryptor.update(data) + data = clt_encryptor.update(plain) + dst_w.write(data) + await dst_w.drain() + except asyncio.CancelledError: + pass + except Exception as e: + log.debug("[%s] forward ended: %s", label, e) + + tasks = [ + asyncio.create_task(forward(reader, remote_writer, True)), + asyncio.create_task(forward(remote_reader, writer, False)), + ] + try: + await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + finally: + for t in tasks: + t.cancel() + for t in tasks: + try: + await t + except BaseException: + pass + for w in (writer, remote_writer): + try: + w.close() + await w.wait_closed() + except BaseException: + pass + + +async def _tcp_fallback(reader, writer, dst, port, relay_init, label, + dc=None, is_media=False, + clt_decryptor=None, clt_encryptor=None, + tg_encryptor=None, tg_decryptor=None): + try: + rr, rw = await asyncio.wait_for( + asyncio.open_connection(dst, port), timeout=10) + except Exception as exc: + log.warning("[%s] TCP fallback to %s:%d failed: %s", + label, dst, port, exc) + return False + + _stats.connections_tcp_fallback += 1 + rw.write(relay_init) + await rw.drain() + await _bridge_tcp_reencrypt(reader, writer, rr, rw, label, + dc=dc, is_media=is_media, + clt_decryptor=clt_decryptor, + clt_encryptor=clt_encryptor, + tg_encryptor=tg_encryptor, + tg_decryptor=tg_decryptor) + return True + + +def _fallback_ip(dc: int) -> Optional[str]: + return DC_DEFAULT_IPS.get(dc) + + +async def _handle_client(reader, writer, secret: bytes): + _stats.connections_total += 1 + peer = writer.get_extra_info('peername') + label = f"{peer[0]}:{peer[1]}" if peer else "?" + + _set_sock_opts(writer.transport) + + try: + try: + handshake = await asyncio.wait_for( + reader.readexactly(HANDSHAKE_LEN), timeout=10) + except asyncio.IncompleteReadError: + log.debug("[%s] client disconnected before handshake", label) + return + + result = _try_handshake(handshake, secret) + if result is None: + _stats.connections_bad += 1 + log.debug("[%s] bad handshake (wrong secret or proto)", label) + try: + while await reader.read(4096): + pass + except Exception: + pass + return + + dc, is_media, proto_tag, client_dec_prekey_iv = result + + if proto_tag == PROTO_TAG_ABRIDGED: + proto_int = PROTO_ABRIDGED_INT + elif proto_tag == PROTO_TAG_INTERMEDIATE: + proto_int = PROTO_INTERMEDIATE_INT + else: + proto_int = PROTO_PADDED_INTERMEDIATE_INT + + dc_idx = -dc if is_media else dc + + log.debug("[%s] handshake ok: DC%d%s proto=0x%08X", + label, dc, ' media' if is_media else '', proto_int) + + relay_init = _generate_relay_init(proto_tag, dc_idx) + + # key = SHA256(prekey + secret), iv from handshake + # "dec" = decrypt data from client; "enc" = encrypt data to client + clt_dec_prekey = client_dec_prekey_iv[:PREKEY_LEN] + clt_dec_iv = client_dec_prekey_iv[PREKEY_LEN:] + clt_dec_key = hashlib.sha256(clt_dec_prekey + secret).digest() + + clt_enc_prekey_iv = client_dec_prekey_iv[::-1] + clt_enc_key = hashlib.sha256( + clt_enc_prekey_iv[:PREKEY_LEN] + secret).digest() + clt_enc_iv = clt_enc_prekey_iv[PREKEY_LEN:] + + clt_decryptor = Cipher( + algorithms.AES(clt_dec_key), modes.CTR(clt_dec_iv) + ).encryptor() + clt_encryptor = Cipher( + algorithms.AES(clt_enc_key), modes.CTR(clt_enc_iv) + ).encryptor() + + # fast-forward client decryptor past the 64-byte init + clt_decryptor.update(ZERO_64) + + # relay side: standard obfuscation (no secret hash, raw key) + relay_enc_key = relay_init[SKIP_LEN:SKIP_LEN + PREKEY_LEN] + relay_enc_iv = relay_init[SKIP_LEN + PREKEY_LEN: + SKIP_LEN + PREKEY_LEN + IV_LEN] + + relay_dec_prekey_iv = relay_init[SKIP_LEN: + SKIP_LEN + PREKEY_LEN + IV_LEN][::-1] + relay_dec_key = relay_dec_prekey_iv[:KEY_LEN] + relay_dec_iv = relay_dec_prekey_iv[KEY_LEN:] + + tg_encryptor = Cipher( + algorithms.AES(relay_enc_key), modes.CTR(relay_enc_iv) + ).encryptor() + tg_decryptor = Cipher( + algorithms.AES(relay_dec_key), modes.CTR(relay_dec_iv) + ).encryptor() + + tg_encryptor.update(ZERO_64) + + dc_key = (dc, is_media) + media_tag = " media" if is_media else "" + + # Fallback if DC not in config or WS blacklisted for this DC/is_media + if dc not in proxy_config.dc_redirects or dc_key in ws_blacklist: + fallback_dst = _fallback_ip(dc) + if fallback_dst: + log.info("[%s] DC%d not in config -> TCP fallback %s:443" + if dc not in proxy_config.dc_redirects else + "[%s] DC%d%s WS blacklisted -> TCP fallback %s:443", + label, dc, fallback_dst) + await _tcp_fallback(reader, writer, fallback_dst, 443, + relay_init, label, dc=dc, + is_media=is_media, + clt_decryptor=clt_decryptor, + clt_encryptor=clt_encryptor, + tg_encryptor=tg_encryptor, + tg_decryptor=tg_decryptor) + else: + log.warning("[%s] DC%d%s no fallback available", + label, dc, media_tag) + return + + now = time.monotonic() + fail_until = dc_fail_until.get(dc_key, 0) + ws_timeout = WS_FAIL_TIMEOUT if now < fail_until else 10.0 + + domains = _ws_domains(dc, is_media) + target = proxy_config.dc_redirects[dc] + ws = None + ws_failed_redirect = False + all_redirects = True + + ws = await _ws_pool.get(dc, is_media, target, domains) + if ws: + log.info("[%s] DC%d%s -> pool hit via %s", + label, dc, media_tag, target) + else: + for domain in domains: + url = f'wss://{domain}/apiws' + log.info("[%s] DC%d%s -> %s via %s", + label, dc, media_tag, url, target) + try: + ws = await RawWebSocket.connect(target, domain, + timeout=ws_timeout) + all_redirects = False + break + except WsHandshakeError as exc: + _stats.ws_errors += 1 + if exc.is_redirect: + ws_failed_redirect = True + log.warning("[%s] DC%d%s got %d from %s -> %s", + label, dc, media_tag, + exc.status_code, domain, + exc.location or '?') + continue + else: + all_redirects = False + log.warning("[%s] DC%d%s WS handshake: %s", + label, dc, media_tag, exc.status_line) + except Exception as exc: + _stats.ws_errors += 1 + all_redirects = False + log.warning("[%s] DC%d%s WS connect failed: %s", + label, dc, media_tag, exc) + + # WS failed -> fallback + if ws is None: + if ws_failed_redirect and all_redirects: + ws_blacklist.add(dc_key) + log.warning("[%s] DC%d%s blacklisted for WS (all 302)", + label, dc, media_tag) + elif ws_failed_redirect: + dc_fail_until[dc_key] = now + DC_FAIL_COOLDOWN + else: + dc_fail_until[dc_key] = now + DC_FAIL_COOLDOWN + log.info("[%s] DC%d%s WS cooldown for %ds", + label, dc, media_tag, int(DC_FAIL_COOLDOWN)) + + fallback_dst = _fallback_ip(dc) or target + log.info("[%s] DC%d%s -> TCP fallback to %s:443", + label, dc, media_tag, fallback_dst) + ok = await _tcp_fallback(reader, writer, fallback_dst, 443, + relay_init, label, dc=dc, + is_media=is_media, + clt_decryptor=clt_decryptor, + clt_encryptor=clt_encryptor, + tg_encryptor=tg_encryptor, + tg_decryptor=tg_decryptor) + if ok: + log.info("[%s] DC%d%s TCP fallback closed", + label, dc, media_tag) + return + + dc_fail_until.pop(dc_key, None) + _stats.connections_ws += 1 + + splitter = None + try: + splitter = _MsgSplitter(relay_init, proto_int) + log.debug("[%s] MsgSplitter activated for proto 0x%08X", + label, proto_int) + except Exception: + pass + + await ws.send(relay_init) + + await _bridge_ws_reencrypt(reader, writer, ws, label, + dc=dc, is_media=is_media, + clt_decryptor=clt_decryptor, + clt_encryptor=clt_encryptor, + tg_encryptor=tg_encryptor, + tg_decryptor=tg_decryptor, + splitter=splitter) + + except asyncio.TimeoutError: + log.warning("[%s] timeout during handshake", label) + except asyncio.IncompleteReadError: + log.debug("[%s] client disconnected", label) + except asyncio.CancelledError: + log.debug("[%s] cancelled", label) + except ConnectionResetError: + log.debug("[%s] connection reset", label) + except OSError as exc: + if getattr(exc, 'winerror', None) == 1236: + log.debug("[%s] connection aborted by local system", label) + else: + log.error("[%s] unexpected OS error: %s", label, exc) + except Exception as exc: + log.error("[%s] unexpected: %s", label, exc.with_traceback()) + finally: + try: + writer.close() + except BaseException: + pass + + +_server_instance = None +_server_stop_event = None + + +async def _run(stop_event: Optional[asyncio.Event] = None): + global _server_instance, _server_stop_event + _server_stop_event = stop_event + + print(proxy_config.secret) + def client_cb(r, w): + asyncio.create_task(_handle_client(r, w, bytes.fromhex(proxy_config.secret))) + + server = await asyncio.start_server(client_cb, proxy_config.host, proxy_config.port) + _server_instance = server + + for sock in server.sockets: + try: + sock.setsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1) + except (OSError, AttributeError): + pass + + link_host = proxy_config.host + if proxy_config.host == '0.0.0.0': + try: + with _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM) as _s: + _s.connect(('8.8.8.8', 80)) + link_host = _s.getsockname()[0] + except OSError: + link_host = '127.0.0.1' + + tg_link = f"tg://proxy?server={link_host}&port={proxy_config.port}&secret=dd{proxy_config.secret}" + + log.info("=" * 60) + log.info(" Telegram MTProto WS Bridge Proxy") + log.info(" Listening on %s:%d", proxy_config.host, proxy_config.port) + log.info(" Secret: %s", "dd" + proxy_config.secret) + log.info(" Target DC IPs:") + for dc in sorted(proxy_config.dc_redirects.keys()): + ip = proxy_config.dc_redirects.get(dc) + log.info(" DC%d: %s", dc, ip) + log.info("=" * 60) + log.info(" Connect link:") + log.info(" %s", tg_link) + log.info("=" * 60) + + async def log_stats(): + try: + while True: + await asyncio.sleep(60) + bl = ', '.join( + f'DC{d}{"m" if m else ""}' + for d, m in sorted(ws_blacklist)) or 'none' + log.info("stats: %s | ws_bl: %s", _stats.summary(), bl) + except asyncio.CancelledError: + raise + + log_stats_task = asyncio.create_task(log_stats()) + + await _ws_pool.warmup(proxy_config.dc_redirects) + + try: + async with server: + if stop_event: + serve_task = asyncio.create_task(server.serve_forever()) + stop_task = asyncio.create_task(stop_event.wait()) + done, _ = await asyncio.wait( + (serve_task, stop_task), + return_when=asyncio.FIRST_COMPLETED, + ) + if stop_task in done: + server.close() + await server.wait_closed() + if not serve_task.done(): + serve_task.cancel() + try: + await serve_task + except asyncio.CancelledError: + pass + else: + stop_task.cancel() + try: + await stop_task + except asyncio.CancelledError: + pass + else: + await server.serve_forever() + finally: + log_stats_task.cancel() + try: + await log_stats_task + except asyncio.CancelledError: + pass + _server_instance = None + + +def parse_dc_ip_list(dc_ip_list: List[str]) -> Dict[int, str]: + dc_redirects: Dict[int, str] = {} + for entry in dc_ip_list: + if ':' not in entry: + raise ValueError( + f"Invalid --dc-ip format {entry!r}, expected DC:IP") + dc_s, ip_s = entry.split(':', 1) + try: + dc_n = int(dc_s) + _socket.inet_aton(ip_s) + except (ValueError, OSError): + raise ValueError(f"Invalid --dc-ip {entry!r}") + dc_redirects[dc_n] = ip_s + return dc_redirects + + +def run_proxy(stop_event: Optional[asyncio.Event] = None): + asyncio.run(_run(stop_event,)) + + +def main(): + ap = argparse.ArgumentParser( + description='Telegram MTProto WebSocket Bridge Proxy') + ap.add_argument('--port', type=int, default=1443, + help=f'Listen port (default 1443)') + ap.add_argument('--host', type=str, default='127.0.0.1', + help='Listen host (default 127.0.0.1)') + ap.add_argument('--secret', type=str, default=None, + help='MTProto proxy secret (32 hex chars). ' + 'Auto-generated if not provided.') + ap.add_argument('--dc-ip', metavar='DC:IP', action='append', + help='Target IP for a DC, e.g. --dc-ip 2:149.154.167.220') + ap.add_argument('-v', '--verbose', action='store_true', + help='Debug logging') + ap.add_argument('--log-file', type=str, default=None, metavar='PATH', + help='Log to file with rotation (default: stderr only)') + ap.add_argument('--log-max-mb', type=float, default=5, metavar='MB', + help='Max log file size in MB before rotation (default 5)') + ap.add_argument('--log-backups', type=int, default=0, metavar='N', + help='Number of rotated log files to keep (default 0)') + ap.add_argument('--buf-kb', type=int, default=256, metavar='KB', + help='Socket send/recv buffer size in KB (default 256)') + ap.add_argument('--pool-size', type=int, default=4, metavar='N', + help='WS connection pool size per DC (default 4, min 0)') + args = ap.parse_args() + + if not args.dc_ip: + args.dc_ip = ['2:149.154.167.220', '4:149.154.167.220'] + + try: + dc_redirects = parse_dc_ip_list(args.dc_ip) + except ValueError as e: + log.error(str(e)) + sys.exit(1) + + if args.secret: + secret_hex = args.secret.strip() + if len(secret_hex) != 32: + log.error("Secret must be exactly 32 hex characters") + sys.exit(1) + try: + secret = bytes.fromhex(secret_hex) + except ValueError: + log.error("Secret must be valid hex") + sys.exit(1) + else: + secret = os.urandom(16).hex() + log.info("Generated secret: %s", secret.hex()) + + global proxy_config + proxy_config = ProxyConfig( + port=args.port, + host=args.host, + secret=secret, + dc_redirects=dc_redirects, + buffer_size=max(4, args.buf_kb) * 1024, + pool_size=max(0, args.pool_size) + ) + + log_level = logging.DEBUG if args.verbose else logging.INFO + log_fmt = logging.Formatter('%(asctime)s %(levelname)-5s %(message)s', + datefmt='%H:%M:%S') + root = logging.getLogger() + root.setLevel(log_level) + + console = logging.StreamHandler() + console.setFormatter(log_fmt) + root.addHandler(console) + + if args.log_file: + fh = logging.handlers.RotatingFileHandler( + args.log_file, + maxBytes=max(32 * 1024, int(args.log_max_mb * 1024 * 1024)), + backupCount=max(0, args.log_backups), + encoding='utf-8', + ) + fh.setFormatter(log_fmt) + root.addHandler(fh) + + try: + asyncio.run(_run(args.port, dc_redirects, secret, host=args.host)) + except KeyboardInterrupt: + log.info("Shutting down. Final stats: %s", _stats.summary()) + + +if __name__ == '__main__': + main() diff --git a/pyproject.toml b/pyproject.toml index 5e440a1..607ecce 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,7 @@ keywords = [ "proxy", "bypass", "websocket", - "socks5", + "mtproto", ] classifiers = [ "Development Status :: 5 - Production/Stable", diff --git a/ui/ctk_theme.py b/ui/ctk_theme.py index 47a3cdd..f814150 100644 --- a/ui/ctk_theme.py +++ b/ui/ctk_theme.py @@ -99,6 +99,30 @@ def create_ctk_root( return root +def create_ctk_toplevel( + ctk: Any, + *, + title: str, + width: int, + height: int, + theme: CtkTheme, + topmost: bool = True, + after_create: Optional[Callable[[Any], None]] = None, +) -> Any: + root = ctk.CTkToplevel() + root.title(title) + root.resizable(False, False) + center_ctk_geometry(root, width, height) + root.configure(fg_color=theme.bg) + if topmost: + root.attributes("-topmost", True) + root.lift() + root.focus_force() + if after_create: + after_create(root) + return root + + def main_content_frame( ctk: Any, root: Any, diff --git a/ui/ctk_tray_ui.py b/ui/ctk_tray_ui.py index fc5b63e..cd26981 100644 --- a/ui/ctk_tray_ui.py +++ b/ui/ctk_tray_ui.py @@ -5,6 +5,7 @@ from __future__ import annotations +import os import webbrowser from dataclasses import dataclass from typing import Any, Callable, Dict, List, Optional, Tuple, Union @@ -22,13 +23,16 @@ from ui.ctk_tooltip import attach_ctk_tooltip, attach_tooltip_to_widgets # Подсказки для формы настроек (новые пользователи) _TIP_HOST = ( - "Адрес, на котором прокси принимает SOCKS5-подключения.\n" + "Адрес, на котором прокси принимает подключения.\n" "Обычно 127.0.0.1 — локальная сеть, 0.0.0.0 - все интерфейсы" ) _TIP_PORT = ( - "Порт SOCKS5. В Telegram Desktop в настройках прокси должен быть " + "Порт прокси. В Telegram Desktop в настройках прокси должен быть " "указан тот же порт" ) +_TIP_SECRET = ( + "Секретный ключ для авторизации клиентов\n" +) _TIP_DC = ( "Соответствие номера датацентра Telegram (DC) и IP-адреса сервера.\n" "Каждая строка: «номер:IP», например 2:149.154.167.220. " @@ -120,6 +124,7 @@ def _config_section( class TrayConfigFormWidgets: host_var: Any port_var: Any + secret_var: Any dc_textbox: Any verbose_var: Any adv_entries: List[Any] @@ -158,7 +163,7 @@ def install_tray_config_form( inner_w = _CONFIG_FORM_INNER_WIDTH - conn = _config_section(ctk, frame, theme, "Подключение SOCKS5") + conn = _config_section(ctk, frame, theme, "Подключение MTProto") host_row = ctk.CTkFrame(conn, fg_color="transparent") host_row.pack(fill="x") @@ -215,6 +220,57 @@ def install_tray_config_form( port_entry.pack(anchor="w") attach_tooltip_to_widgets([port_lbl, port_entry, port_col], _TIP_PORT) + secret_row = ctk.CTkFrame(conn, fg_color="transparent") + secret_row.pack(fill="x") + + secret_col = ctk.CTkFrame(secret_row, fg_color="transparent") + secret_col.pack(side="left", fill="x", expand=True, padx=(0, 10)) + secret_lbl = ctk.CTkLabel( + secret_col, + text="Secret", + font=(theme.ui_font_family, 12), + text_color=theme.text_secondary, + anchor="w", + ) + secret_lbl.pack(anchor="w", pady=(0, 2)) + secret_var = ctk.StringVar(value=cfg.get("secret", default_config["secret"])) + secret_entry = ctk.CTkEntry( + secret_col, + textvariable=secret_var, + width=160, + height=36, + font=(theme.ui_font_family, 13), + corner_radius=10, + fg_color=theme.bg, + border_color=theme.field_border, + border_width=1, + text_color=theme.text_primary, + ) + secret_entry.pack(fill="x", pady=(0, 0)) + attach_tooltip_to_widgets([secret_lbl, secret_entry, secret_col], _TIP_SECRET) + + regen_col = ctk.CTkFrame(secret_row, fg_color="transparent") + regen_col.pack(side="left", anchor="s") + ctk.CTkLabel( + regen_col, + text="", + font=(theme.ui_font_family, 12), + ).pack(pady=(0, 2)) + ctk.CTkButton( + regen_col, + text="↺", + width=36, + height=36, + font=(theme.ui_font_family, 18), + corner_radius=10, + fg_color=theme.tg_blue, + hover_color=theme.tg_blue_hover, + text_color="#ffffff", + border_width=1, + border_color=theme.field_border, + command=lambda: secret_var.set(os.urandom(16).hex()), + ).pack() + dc_inner = _config_section(ctk, frame, theme, "Датацентры Telegram (DC → IP)") dc_lbl = ctk.CTkLabel( dc_inner, @@ -395,6 +451,7 @@ def install_tray_config_form( return TrayConfigFormWidgets( host_var=host_var, port_var=port_var, + secret_var=secret_var, dc_textbox=dc_textbox, verbose_var=verbose_var, adv_entries=adv_entries, @@ -459,6 +516,7 @@ def validate_config_form( new_cfg: Dict[str, Any] = { "host": host_val, "port": port_val, + "secret": widgets.secret_var.get().strip(), "dc_ip": lines, "verbose": widgets.verbose_var.get(), } @@ -517,12 +575,13 @@ def populate_first_run_window( *, host: str, port: int, + secret: str, on_done: Callable[[bool], None], ) -> None: """ Содержимое окна первого запуска. on_done(open_in_telegram) — по «Начать» и по закрытию окна. """ - tg_url = f"tg://socks?server={host}&port={port}" + tg_url = f"tg://proxy?server={host}&port={port}&secret=dd{secret}" fpx, fpy = FIRST_RUN_FRAME_PAD frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy) @@ -544,7 +603,8 @@ def populate_first_run_window( (f" Или ссылка: {tg_url}", False), ("\n Вручную:", True), (" Настройки → Продвинутые → Тип подключения → Прокси", False), - (f" SOCKS5 → {host} : {port} (без логина/пароля)", False), + (f" MTProto → {host} : {port}", False), + (f" Secret: dd{secret}", False), ] for text, bold in sections: diff --git a/utils/default_config.py b/utils/default_config.py index 30b7bc6..c1152a9 100644 --- a/utils/default_config.py +++ b/utils/default_config.py @@ -5,10 +5,11 @@ from __future__ import annotations import sys +import os from typing import Any, Dict _TRAY_DEFAULTS_COMMON: Dict[str, Any] = { - "port": 1080, + "port": 1443, "host": "127.0.0.1", "dc_ip": ["2:149.154.167.220", "4:149.154.167.220"], "verbose": False, @@ -22,6 +23,9 @@ _TRAY_DEFAULTS_COMMON: Dict[str, Any] = { def default_tray_config() -> Dict[str, Any]: """Новая копия конфига по умолчанию для текущей ОС.""" cfg = dict(_TRAY_DEFAULTS_COMMON) + cfg["secret"] = os.urandom(16).hex() + if sys.platform == "win32": cfg["autostart"] = False + return cfg diff --git a/windows.py b/windows.py index 4357fe0..c140d81 100644 --- a/windows.py +++ b/windows.py @@ -37,7 +37,9 @@ except ImportError: Image = ImageDraw = ImageFont = None import proxy.tg_ws_proxy as tg_ws_proxy +from proxy.tg_ws_proxy import proxy_config from proxy import __version__ + from utils.default_config import default_tray_config from ui.ctk_tray_ui import ( install_tray_config_buttons, @@ -50,7 +52,7 @@ from ui.ctk_theme import ( CONFIG_DIALOG_FRAME_PAD, CONFIG_DIALOG_SIZE, FIRST_RUN_SIZE, - create_ctk_root, + create_ctk_toplevel, ctk_theme_for_platform, main_content_frame, ) @@ -76,6 +78,9 @@ _config: dict = {} _exiting: bool = False _lock_file_path: Optional[Path] = None +_ctk_root = None +_ctk_root_ready = threading.Event() + log = logging.getLogger("tg-ws-tray") _user32 = ctypes.windll.user32 @@ -312,17 +317,18 @@ def _load_icon(): -def _run_proxy_thread(port: int, dc_opt: Dict[int, str], verbose: bool, - host: str = '127.0.0.1'): +def _run_proxy_thread(): global _async_stop + loop = _asyncio.new_event_loop() _asyncio.set_event_loop(loop) + stop_ev = _asyncio.Event() _async_stop = (loop, stop_ev) try: loop.run_until_complete( - tg_ws_proxy._run(port, dc_opt, stop_event=stop_ev, host=host)) + tg_ws_proxy._run(stop_event=stop_ev)) except Exception as exc: log.error("Proxy thread crashed: %s", exc) if "10048" in str(exc) or "Address already in use" in str(exc): @@ -341,27 +347,29 @@ def start_proxy(): cfg = _config port = cfg.get("port", DEFAULT_CONFIG["port"]) host = cfg.get("host", DEFAULT_CONFIG["host"]) + secret = cfg.get("secret", DEFAULT_CONFIG["secret"]) dc_ip_list = cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"]) - verbose = cfg.get("verbose", False) + buf_kb = cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"]) + pool_size = cfg.get("pool_size", DEFAULT_CONFIG["pool_size"]) try: - dc_opt = tg_ws_proxy.parse_dc_ip_list(dc_ip_list) + dc_redirects = tg_ws_proxy.parse_dc_ip_list(dc_ip_list) except ValueError as e: log.error("Bad config dc_ip: %s", e) _show_error(f"Ошибка конфигурации:\n{e}") return - log.info("Starting proxy on %s:%d ...", host, port) + proxy_config.port = port + proxy_config.host = host + proxy_config.secret = secret + proxy_config.dc_redirects = dc_redirects + proxy_config.buffer_size = max(4, buf_kb) * 1024 + proxy_config.pool_size = max(0, pool_size) - buf_kb = cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"]) - pool_size = cfg.get("pool_size", DEFAULT_CONFIG["pool_size"]) - tg_ws_proxy._RECV_BUF = max(4, buf_kb) * 1024 - tg_ws_proxy._SEND_BUF = tg_ws_proxy._RECV_BUF - tg_ws_proxy._WS_POOL_SIZE = max(0, pool_size) + log.info("Starting proxy on %s:%d ...", host, port) _proxy_thread = threading.Thread( target=_run_proxy_thread, - args=(port, dc_opt, verbose, host), daemon=True, name="proxy") _proxy_thread.start() @@ -440,10 +448,55 @@ def _maybe_notify_update_async(): threading.Thread(target=_work, daemon=True, name="update-check").start() +def _ensure_ctk_thread() -> bool: + """Start the persistent hidden CTk root in its own thread (once).""" + global _ctk_root + if ctk is None: + return False + if _ctk_root_ready.is_set(): + return True + + def _run(): + global _ctk_root + from ui.ctk_theme import ( + apply_ctk_appearance, + _install_tkinter_variable_del_guard, + ) + _install_tkinter_variable_del_guard() + apply_ctk_appearance(ctk) + _ctk_root = ctk.CTk() + _ctk_root.withdraw() + _ctk_root_ready.set() + _ctk_root.mainloop() + + threading.Thread(target=_run, daemon=True, name="ctk-root").start() + _ctk_root_ready.wait(timeout=5.0) + return _ctk_root is not None + + +def _ctk_run_dialog(build_fn) -> None: + """Schedule build_fn(done_event) on the CTk thread and block until done_event is set.""" + if _ctk_root is None: + return + done = threading.Event() + + def _invoke(): + try: + build_fn(done) + except Exception: + log.exception("CTk dialog failed") + done.set() + + _ctk_root.after(0, _invoke) + done.wait() + + def _on_open_in_telegram(icon=None, item=None): host = _config.get("host", DEFAULT_CONFIG["host"]) port = _config.get("port", DEFAULT_CONFIG["port"]) - url = f"tg://socks?server={host}&port={port}" + secret = _config.get("secret", DEFAULT_CONFIG["secret"]) + + url = f"tg://proxy?server={host}&port={port}&secret=dd{secret}" log.info("Opening %s", url) try: result = webbrowser.open(url) @@ -476,96 +529,84 @@ def _on_edit_config(icon=None, item=None): def _edit_config_dialog(): - if ctk is None: + if not _ensure_ctk_thread(): _show_error("customtkinter не установлен.") return cfg = dict(_config) cfg["autostart"] = is_autostart_enabled() - - # Make sure that the autostart key is removed if autostart - # is disabled, even if the executable file is moved. if _supports_autostart() and not cfg["autostart"]: set_autostart_enabled(False) - theme = ctk_theme_for_platform() - w, h = CONFIG_DIALOG_SIZE - if _supports_autostart(): - h += 100 - - icon_path = str(Path(__file__).parent / "icon.ico") - - root = create_ctk_root( - ctk, - title="TG WS Proxy — Настройки", - width=w, - height=h, - theme=theme, - after_create=lambda r: r.iconbitmap(icon_path), - ) - - fpx, fpy = CONFIG_DIALOG_FRAME_PAD - frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy) - - scroll, footer = tray_settings_scroll_and_footer(ctk, frame, theme) - - widgets = install_tray_config_form( - ctk, - scroll, - theme, - cfg, - DEFAULT_CONFIG, - show_autostart=_supports_autostart(), - autostart_value=cfg.get("autostart", False), - ) - - def on_save(): - merged = validate_config_form( - widgets, - DEFAULT_CONFIG, - include_autostart=_supports_autostart(), - ) - if isinstance(merged, str): - _show_error(merged) - return - - new_cfg = merged - save_config(new_cfg) - _config.update(new_cfg) - log.info("Config saved: %s", new_cfg) - + def _build(done: threading.Event): + theme = ctk_theme_for_platform() + w, h = CONFIG_DIALOG_SIZE if _supports_autostart(): - set_autostart_enabled(bool(new_cfg.get("autostart", False))) + h += 100 - _tray_icon.menu = _build_menu() + icon_path = str(Path(__file__).parent / "icon.ico") + root = create_ctk_toplevel( + ctk, + title="TG WS Proxy — Настройки", + width=w, + height=h, + theme=theme, + after_create=lambda r: r.iconbitmap(icon_path), + ) - # Win32 MessageBox из того же потока, что и mainloop CTk, блокирует обработку Tcl/Tk - # и даёт зависание; tkinter.messagebox согласован с циклом окна. - from tkinter import messagebox - if messagebox.askyesno("Перезапустить?", - "Настройки сохранены.\n\n" - "Перезапустить прокси сейчас?", - parent=root): - root.destroy() - restart_proxy() - else: + fpx, fpy = CONFIG_DIALOG_FRAME_PAD + frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy) + scroll, footer = tray_settings_scroll_and_footer(ctk, frame, theme) + widgets = install_tray_config_form( + ctk, + scroll, + theme, + cfg, + DEFAULT_CONFIG, + show_autostart=_supports_autostart(), + autostart_value=cfg.get("autostart", False), + ) + + def _finish(): root.destroy() + done.set() - def on_cancel(): - root.destroy() + def on_save(): + merged = validate_config_form( + widgets, + DEFAULT_CONFIG, + include_autostart=_supports_autostart(), + ) + if isinstance(merged, str): + _show_error(merged) + return - install_tray_config_buttons( - ctk, footer, theme, on_save=on_save, on_cancel=on_cancel) + new_cfg = merged + save_config(new_cfg) + _config.update(new_cfg) + log.info("Config saved: %s", new_cfg) + if _supports_autostart(): + set_autostart_enabled(bool(new_cfg.get("autostart", False))) + _tray_icon.menu = _build_menu() - try: - root.mainloop() - finally: - import tkinter as tk - try: - if root.winfo_exists(): - root.destroy() - except tk.TclError: - pass + from tkinter import messagebox + do_restart = messagebox.askyesno( + "Перезапустить?", + "Настройки сохранены.\n\nПерезапустить прокси сейчас?", + parent=root) + _finish() + if do_restart: + threading.Thread( + target=restart_proxy, daemon=True).start() + + def on_cancel(): + _finish() + + root.protocol("WM_DELETE_WINDOW", on_cancel) + install_tray_config_buttons( + ctk, footer, theme, on_save=on_save, on_cancel=on_cancel) + + _ctk_run_dialog(_build) def _on_open_logs(icon=None, item=None): @@ -584,6 +625,12 @@ def _on_exit(icon=None, item=None): _exiting = True log.info("User requested exit") + if _ctk_root is not None: + try: + _ctk_root.after(0, _ctk_root.quit) + except Exception: + pass + def _force_exit(): time.sleep(3) os._exit(0) @@ -593,49 +640,43 @@ def _on_exit(icon=None, item=None): icon.stop() - def _show_first_run(): _ensure_dirs() if FIRST_RUN_MARKER.exists(): return - - host = _config.get("host", DEFAULT_CONFIG["host"]) - port = _config.get("port", DEFAULT_CONFIG["port"]) - - if ctk is None: + if not _ensure_ctk_thread(): FIRST_RUN_MARKER.touch() return - theme = ctk_theme_for_platform() - icon_path = str(Path(__file__).parent / "icon.ico") - w, h = FIRST_RUN_SIZE - root = create_ctk_root( - ctk, - title="TG WS Proxy", - width=w, - height=h, - theme=theme, - after_create=lambda r: r.iconbitmap(icon_path), - ) + host = _config.get("host", DEFAULT_CONFIG["host"]) + port = _config.get("port", DEFAULT_CONFIG["port"]) + secret = _config.get("secret", DEFAULT_CONFIG["secret"]) - def on_done(open_tg: bool): - FIRST_RUN_MARKER.touch() - root.destroy() - if open_tg: - _on_open_in_telegram() + def _build(done: threading.Event): + theme = ctk_theme_for_platform() + icon_path = str(Path(__file__).parent / "icon.ico") + w, h = FIRST_RUN_SIZE + root = create_ctk_toplevel( + ctk, + title="TG WS Proxy", + width=w, + height=h, + theme=theme, + after_create=lambda r: r.iconbitmap(icon_path), + ) - populate_first_run_window( - ctk, root, theme, host=host, port=port, on_done=on_done) + def on_done(open_tg: bool): + FIRST_RUN_MARKER.touch() + root.destroy() + done.set() + if open_tg: + _on_open_in_telegram() - try: - root.mainloop() - finally: - import tkinter as tk - try: - if root.winfo_exists(): - root.destroy() - except tk.TclError: - pass + populate_first_run_window( + ctk, root, theme, host=host, port=port, secret=secret, + on_done=on_done) + + _ctk_run_dialog(_build) def _has_ipv6_enabled() -> bool: @@ -695,9 +736,11 @@ def _build_menu(): return None host = _config.get("host", DEFAULT_CONFIG["host"]) port = _config.get("port", DEFAULT_CONFIG["port"]) + link_host = tg_ws_proxy.get_link_host(host) + return pystray.Menu( pystray.MenuItem( - f"Открыть в Telegram ({host}:{port})", + f"Открыть в Telegram ({link_host}:{port})", _on_open_in_telegram, default=True), pystray.Menu.SEPARATOR,