docs update

.github/cfproxy-domains.txt

@@ -0,0 +1,2 @@
+virkgj.com
+vmmzovy.com

.github/workflows/build.yml

@@ -29,15 +29,43 @@ jobs:
           python-version: "3.12"
           cache: "pip"
 
+      - name: Setup MSVC 14.40 toolset
+        uses: ilammy/msvc-dev-cmd@v1
+        with:
+          toolset: 14.40
+
       - name: Install dependencies
         run: pip install .
 
-      - name: Install pyinstaller
+      - name: Build PyInstaller bootloader from source
-        run: pip install "pyinstaller==6.13.0"
+        run: |
+          pip install "pyinstaller==6.16.0" --no-binary pyinstaller
+        env:
+          PYINSTALLER_COMPILE_BOOTLOADER: 1
 
       - name: Build EXE with PyInstaller
         run: pyinstaller packaging/windows.spec --noconfirm
 
+      - name: Strip Rich PE header
+        shell: bash
+        run: |
+          python -c "
+          import struct, pathlib
+          exe = pathlib.Path('dist/TgWsProxy.exe')
+          data = bytearray(exe.read_bytes())
+          rich = data.find(b'Rich')
+          if rich == -1:
+              raise SystemExit('Rich header not found')
+          ck = struct.unpack_from('<I', data, rich + 4)[0]
+          dans = struct.pack('<I', 0x536E6144 ^ ck)
+          ds = data.find(dans)
+          if ds == -1:
+              raise SystemExit('DanS marker not found')
+          data[ds:rich + 8] = b'\x00' * (rich + 8 - ds)
+          exe.write_bytes(data)
+          print(f'Stripped Rich header: offset {ds}..{rich+8}')
+          "
+
       - name: Rename artifact
         run: mv dist/TgWsProxy.exe dist/TgWsProxy_windows.exe
 
@@ -71,6 +99,26 @@ jobs:
       - name: Build EXE with PyInstaller
         run: pyinstaller packaging/windows.spec --noconfirm
 
+      - name: Strip Rich PE header
+        shell: bash
+        run: |
+          python -c "
+          import struct, pathlib
+          exe = pathlib.Path('dist/TgWsProxy.exe')
+          data = bytearray(exe.read_bytes())
+          rich = data.find(b'Rich')
+          if rich == -1:
+              raise SystemExit('Rich header not found')
+          ck = struct.unpack_from('<I', data, rich + 4)[0]
+          dans = struct.pack('<I', 0x536E6144 ^ ck)
+          ds = data.find(dans)
+          if ds == -1:
+              raise SystemExit('DanS marker not found')
+          data[ds:rich + 8] = b'\x00' * (rich + 8 - ds)
+          exe.write_bytes(data)
+          print(f'Stripped Rich header: offset {ds}..{rich+8}')
+          "
+
       - name: Rename artifact
         run: mv dist/TgWsProxy.exe dist/TgWsProxy_windows_7_${{ matrix.suffix }}.exe
 
@@ -145,7 +193,7 @@ jobs:
 
           python3.12 -m pip install --no-deps wheelhouse/universal2/*.whl
           python3.12 -m pip install .
-          python3.12 -m pip install pyinstaller==6.13.0
+          python3.12 -m pip install pyinstaller==6.16.0
 
       - name: Create macOS icon from ICO
         run: |
@@ -247,7 +295,7 @@ jobs:
         run: |
           .venv/bin/pip install --upgrade pip
           .venv/bin/pip install .
-          .venv/bin/pip install "pyinstaller==6.13.0"
+          .venv/bin/pip install "pyinstaller==6.16.0"
 
       - name: Build binary with PyInstaller
         run: .venv/bin/pyinstaller packaging/linux.spec --noconfirm

Dockerfile

@@ -35,7 +35,7 @@ RUN apt-get update \
 WORKDIR /app
 COPY --from=builder /opt/venv /opt/venv
 COPY proxy ./proxy
-COPY README.md LICENSE ./
+COPY docs/README.md LICENSE ./
 
 USER app
 

docs/CfProxy.md

@@ -0,0 +1,29 @@
+# Cloudflare Proxy
+
+Для недоступных датацентров можно использовать альтернативный бесплатный метод подключения - проксирование через Cloudflare. **Для работы нужен только домен**. В приложении есть домен по умолчанию, но его можно (и лучше) заменить на свой.
+  
+Прокси возвращает доступ к тому, что до этого не грузило (реакциям, некоторым стикерам). Если у вас до этого не грузило видео/фото на аккаунте без премиума, то уберите всё кроме `4:149.154.167.220` из `DC->IP` блока в настройках. Если CF-прокси у вас работает - медиа снова начнёт грузиться.
+
+## Зачем мне настраивать свой домен?
+Cloudflare имеет лимиты на одновременное количество подключений WS. Домен по умолчанию может перестать работать в любой момент. 
+
+## Настройка своего домена
+1. Добавьте свой домен в Cloudflare (либо купив у них напрямую, либо поменяв NS сервера: https://developers.cloudflare.com/dns/zone-setups/full-setup/setup/). Домены стоят +- 150 рублей на год, подойдёт любой.
+
+2. В `SSL/TLS` -> `Overview` выставьте режим **Flexible**
+
+3. В `DNS` -> `Records` добавьте следующие `A` записи через `+ Add Record`:
+- Name=`kws1`   IPv4=`149.154.175.50`
+- Name=`kws2`   IPv4=`149.154.167.51`
+- Name=`kws3`   IPv4=`149.154.175.100`
+- Name=`kws4`   IPv4=`149.154.167.91`
+- Name=`kws5`   IPv4=`149.154.171.5`
+- Name=`kws203` IPv4=`91.105.192.100`
+
+4. **Добавьте домен в [zapret](https://github.com/Flowseal/zapret-discord-youtube/) или другой софт для обхода блокировок, так как подсеть Cloudflare забанена (по крайней мере, если вы из России)**
+
+5. В настройках TgWsProxy поменяйте домен на свой
+
+## Mentions
+Idea - https://github.com/Nekogram/WSProxy  
+Thanks to [@UjuiUjuMandan](https://github.com/UjuiUjuMandan) for the information

docs/README.md

@@ -1,3 +1,11 @@
+> [!TIP]
+>
+> ### 🎉 Поддержать меня
+>
+> USDT (TRC20): `TXPnKs2Ww1RD8JN6nChFUVmi5r2hqrWjuu`  
+> BTC: `bc1qr8vd6jelkyyry3m4mq6z5txdx4pl856fu6ss0w`  
+> ETH: `0x1417878fdc5047E670a77748B34819b9A49C72F1`  
+
 > [!CAUTION]
 >
 > ### Реакция антивирусов
@@ -26,7 +34,13 @@ Telegram Desktop → MTProto Proxy (127.0.0.1:1443) → WebSocket → Telegram D
 2. Перехватывает подключения к IP-адресам Telegram
 3. Извлекает DC ID из MTProto obfuscation init-пакета
 4. Устанавливает WebSocket (TLS) соединение к соответствующему DC через домены Telegram
-5. Если WS недоступен (302 redirect) — автоматически переключается на прямое TCP-соединение
+5. Если WS недоступен (302 redirect) — автоматически переключается на CfProxy / прямое TCP-соединение
+
+> [!IMPORTANT] 
+> ### Не грузит фото/видео?
+> ### Удалите в настройках прокси в DC->IP всё, кроме `4:149.154.167.220`
+> Подобная проблема встречается на аккаунтах без Premium  
+> Если вам не помогло, то настраивайте свой домен по гайду отсюда: https://github.com/Flowseal/tg-ws-proxy/blob/main/docs/CfProxy.md
 
 ## 🚀 Быстрый старт
 
@@ -69,8 +83,9 @@ makepkg -si
 # При помощи AUR-helper
 paru -S tg-ws-proxy-bin
 
-# Если вы установили -cli пакет, то запуск осуществляется через systemctl, где 8888 это номер порта прокси:
+# Если вы установили -cli пакет, то запуск осуществляется через systemctl, где 8888 это номер порта,
-sudo systemctl start tg-ws-proxy-cli@8888
+# разделитель ":" и secret, который можно сгенерировать командой: openssl rand -hex 16
+sudo systemctl start tg-ws-proxy-cli@8888:3075abe65830f0325116bb0416cadf9f
 ```
 
 Для остальных дистрибутивов можно использовать **`TgWsProxy_linux_amd64`** (бинарный файл для x86_64).
@@ -128,11 +143,14 @@ tg-ws-proxy [--port PORT] [--host HOST] [--dc-ip DC:IP ...] [-v]
 | `--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` | Размер буфера в КБ
+| `--no-cfproxy` | `false` | Отключить попытку [проксирования через Cloudflare]((https://github.com/Flowseal/tg-ws-proxy/blob/main/docs/CfProxy.md)) |
-| `--pool-size` | `4` | Количество заготовленных соединений на каждый DC
+| `--cfproxy-domain` | | Указать свой домен для проксирования через Cloudfalre. [Подробнее тут](https://github.com/Flowseal/tg-ws-proxy/blob/main/docs/CfProxy.md) |
-| `--log-file` | выкл. | Путь до файла, в который сохранять логи 
+| `--cfproxy-priority` | `true` | Пробовать проксировать через Cloudflare перед прямым TCP подключением |
-| `--log-max-mb` | `5` | Максимальный размер файла логов в МБ (после идёт перезапись)
+| `--buf-kb` | `256` | Размер буфера в КБ |
-| `--log-backups` | `0` | Количество сохранений логов после перезаписи
+| `--pool-size` | `4` | Количество заготовленных соединений на каждый DC |
+| `--log-file` | выкл. | Путь до файла, в который сохранять логи  |
+| `--log-max-mb` | `5` | Максимальный размер файла логов в МБ (после идёт перезапись) |
+| `--log-backups` | `0` | Количество сохранений логов после перезаписи |
 | `-v`, `--verbose` | выкл. | Подробное логирование (DEBUG) |
 
 **Примеры:**
@@ -148,20 +166,6 @@ tg-ws-proxy --port 9050 --dc-ip 1:149.154.175.205 --dc-ip 2:149.154.167.220
 tg-ws-proxy -v
 ```
 
-## CLI-скрипты (pyproject.toml)
-
-CLI команды объявляются в `pyproject.toml` в секции `[project.scripts]` и должны указывать на `module:function`.
-
-Пример:
-
-```toml
-[project.scripts]
-tg-ws-proxy = "proxy.tg_ws_proxy:main"
-tg-ws-proxy-tray-win = "windows:main"
-tg-ws-proxy-tray-macos = "macos:main"
-tg-ws-proxy-tray-linux = "linux:main"
-```
-
 ## Настройка Telegram Desktop
 
 ### Автоматически

linux.py

@@ -12,7 +12,7 @@ import pyperclip
 import pystray
 from PIL import Image, ImageTk
 
-import proxy.tg_ws_proxy as tg_ws_proxy
+from proxy import get_link_host
 
 from utils.tray_common import (
     APP_NAME, DEFAULT_CONFIG, FIRST_RUN_MARKER, LOG_FILE,
@@ -138,7 +138,7 @@ def _on_exit(icon=None, item=None) -> None:
 
 
 def _edit_config_dialog() -> None:
-    if not ensure_ctk_thread(ctk):
+    if not ensure_ctk_thread(ctk, _config.get("appearance", "auto")):
         _show_error("customtkinter не установлен.")
         return
 
@@ -193,7 +193,7 @@ def _show_first_run() -> None:
     ensure_dirs()
     if FIRST_RUN_MARKER.exists():
         return
-    if not ensure_ctk_thread(ctk):
+    if not ensure_ctk_thread(ctk, _config.get("appearance", "auto")):
         FIRST_RUN_MARKER.touch()
         return
 
@@ -227,7 +227,7 @@ def _show_first_run() -> None:
 def _build_menu():
     host = _config.get("host", DEFAULT_CONFIG["host"])
     port = _config.get("port", DEFAULT_CONFIG["port"])
-    link_host = tg_ws_proxy.get_link_host(host)
+    link_host = get_link_host(host)
     return pystray.Menu(
         pystray.MenuItem(f"Открыть в Telegram ({link_host}:{port})", _on_open_in_telegram, default=True),
         pystray.MenuItem("Скопировать ссылку", _on_copy_link),

macos.py

@@ -24,8 +24,8 @@ try:
 except ImportError:
     pyperclip = None
 
-import proxy.tg_ws_proxy as tg_ws_proxy
+from proxy import __version__, get_link_host, parse_dc_ip_list, proxy_config
-from proxy import __version__
+from proxy.tg_ws_proxy import _run
 
 from utils.tray_common import (
     APP_DIR, APP_NAME, DEFAULT_CONFIG, FIRST_RUN_MARKER, IPV6_WARN_MARKER,
@@ -153,7 +153,7 @@ def _run_proxy_thread() -> None:
     stop_ev = _asyncio.Event()
     _async_stop = (loop, stop_ev)
     try:
-        loop.run_until_complete(tg_ws_proxy._run(stop_event=stop_ev))
+        loop.run_until_complete(_run(stop_event=stop_ev))
     except Exception as exc:
         log.error("Proxy thread crashed: %s", exc)
         if "Address already in use" in str(exc):
@@ -176,7 +176,7 @@ def _start_proxy() -> None:
     if not apply_proxy_config(_config):
         _show_error("Ошибка конфигурации DC → IP.")
         return
-    pc = tg_ws_proxy.proxy_config
+    pc = proxy_config
     log.info("Starting proxy on %s:%d ...", pc.host, pc.port)
     _proxy_thread = threading.Thread(target=_run_proxy_thread, daemon=True, name="proxy")
     _proxy_thread.start()
@@ -362,7 +362,7 @@ def _edit_config_dialog() -> None:
         return
     dc_lines = [s.strip() for s in dc_str.replace(",", "\n").splitlines() if s.strip()]
     try:
-        tg_ws_proxy.parse_dc_ip_list(dc_lines)
+        parse_dc_ip_list(dc_lines)
     except ValueError as e:
         _show_error(str(e))
         return
@@ -392,6 +392,26 @@ def _edit_config_dialog() -> None:
                 except ValueError:
                     pass
 
+    cfproxy = _ask_yes_no_close("Включить Cloudflare Proxy (CfProxy)?")
+    if cfproxy is None:
+        return
+
+    cfproxy_priority = True
+    if cfproxy:
+        cfproxy_priority_result = _ask_yes_no_close("Приоритет CfProxy (пробовать раньше прямого TCP)?")
+        if cfproxy_priority_result is None:
+            return
+        cfproxy_priority = cfproxy_priority_result
+
+    cfproxy_domain = _osascript_input(
+        "Свой CF-домен (оставьте пустым для автоматического выбора):\n"
+        "DNS записи kws1-kws5,kws203 должны указывать на IP датацентров Telegram через Cloudflare.",
+        cfg.get("cfproxy_user_domain", DEFAULT_CONFIG.get("cfproxy_user_domain", "")),
+    )
+    if cfproxy_domain is None:
+        return
+    cfproxy_domain = cfproxy_domain.strip()
+
     new_cfg = {
         "host": host,
         "port": port,
@@ -402,6 +422,9 @@ def _edit_config_dialog() -> None:
         "pool_size": adv.get("pool_size", cfg.get("pool_size", DEFAULT_CONFIG["pool_size"])),
         "log_max_mb": adv.get("log_max_mb", cfg.get("log_max_mb", DEFAULT_CONFIG["log_max_mb"])),
         "check_updates": cfg.get("check_updates", True),
+        "cfproxy": cfproxy,
+        "cfproxy_priority": cfproxy_priority,
+        "cfproxy_user_domain": cfproxy_domain,
     }
     save_config(new_cfg)
     log.info("Config saved: %s", new_cfg)
@@ -427,7 +450,7 @@ def _show_first_run() -> None:
     port = _config.get("port", DEFAULT_CONFIG["port"])
     secret = _config.get("secret", DEFAULT_CONFIG["secret"])
     tg_url = tg_proxy_url(_config)
-    link_host = tg_ws_proxy.get_link_host(host)
+    link_host = get_link_host(host)
 
     text = (
         f"Прокси запущен и работает в строке меню.\n\n"
@@ -496,7 +519,7 @@ 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)
+        link_host = get_link_host(host)
 
         self._open_tg_item = rumps.MenuItem(
             f"Открыть в Telegram ({link_host}:{port})", callback=_on_open_in_telegram
@@ -536,7 +559,7 @@ class TgWsProxyApp(_TgWsProxyAppBase):
     def update_menu_title(self) -> None:
         host = _config.get("host", DEFAULT_CONFIG["host"])
         port = _config.get("port", DEFAULT_CONFIG["port"])
-        link_host = tg_ws_proxy.get_link_host(host)
+        link_host = get_link_host(host)
         self._open_tg_item.title = f"Открыть в Telegram ({link_host}:{port})"
 
 

packaging/linux.spec

@@ -46,11 +46,25 @@ a = Analysis(
     hookspath=[],
     hooksconfig={},
     runtime_hooks=[],
-    excludes=[],
+    excludes=[
+        'PIL._avif',
+        'PIL._webp',
+        'PIL._imagingtk',
+    ],
     noarchive=False,
     cipher=block_cipher,
 )
 
+_PIL_EXCLUDE_PYDS = {
+    '_avif', '_webp', '_imagingtk',
+    'FpxImagePlugin', 'MicImagePlugin',
+}
+a.binaries = [
+    (name, path, typ)
+    for name, path, typ in a.binaries
+    if not any(ex in name for ex in _PIL_EXCLUDE_PYDS)
+]
+
 icon_path = os.path.join(os.path.dirname(SPEC), os.pardir, 'icon.ico')
 if os.path.exists(icon_path):
     a.datas += [('icon.ico', icon_path, 'DATA')]

packaging/macos.spec

@@ -25,11 +25,25 @@ a = Analysis(
     hookspath=[],
     hooksconfig={},
     runtime_hooks=[],
-    excludes=[],
+    excludes=[
+        'PIL._avif',
+        'PIL._webp',
+        'PIL._imagingtk',
+    ],
     noarchive=False,
     cipher=block_cipher,
 )
 
+_PIL_EXCLUDE_PYDS = {
+    '_avif', '_webp', '_imagingtk',
+    'FpxImagePlugin', 'MicImagePlugin',
+}
+a.binaries = [
+    (name, path, typ)
+    for name, path, typ in a.binaries
+    if not any(ex in name for ex in _PIL_EXCLUDE_PYDS)
+]
+
 icon_path = os.path.join(os.path.dirname(SPEC), os.pardir, 'icon.icns')
 if not os.path.exists(icon_path):
     icon_path = None

packaging/version_info.txt

@@ -0,0 +1,36 @@
+# UTF-8
+#
+# For more details about fixed file info 'ffi' see:
+# http://msdn.microsoft.com/en-us/library/ms646997.aspx
+VSVersionInfo(
+  ffi=FixedFileInfo(
+    filevers=(1, 0, 0, 0),
+    prodvers=(1, 0, 0, 0),
+    mask=0x3f,
+    flags=0x0,
+    OS=0x40004,
+    fileType=0x1,
+    subtype=0x0,
+    date=(0, 0)
+  ),
+  kids=[
+    StringFileInfo(
+      [
+        StringTable(
+          u'040904B0',
+          [
+            StringStruct(u'CompanyName', u'Flowseal'),
+            StringStruct(u'FileDescription', u'Telegram Desktop WebSocket Bridge Proxy'),
+            StringStruct(u'FileVersion', u'1.0.0.0'),
+            StringStruct(u'InternalName', u'TgWsProxy'),
+            StringStruct(u'LegalCopyright', u'Copyright (c) Flowseal. MIT License.'),
+            StringStruct(u'OriginalFilename', u'TgWsProxy.exe'),
+            StringStruct(u'ProductName', u'TG WS Proxy'),
+            StringStruct(u'ProductVersion', u'1.0.0.0'),
+          ]
+        )
+      ]
+    ),
+    VarFileInfo([VarStruct(u'Translation', [1033, 1200])])
+  ]
+)

packaging/windows.spec

@@ -26,14 +26,29 @@ a = Analysis(
     hookspath=[],
     hooksconfig={},
     runtime_hooks=[],
-    excludes=[],
+    excludes=[
+        'PIL._avif',
+        'PIL._webp',
+        'PIL._imagingtk',
+    ],
     win_no_prefer_redirects=False,
     win_private_assemblies=False,
     cipher=block_cipher,
     noarchive=False,
 )
 
+_PIL_EXCLUDE_PYDS = {
+    '_avif', '_webp', '_imagingtk',
+    'FpxImagePlugin', 'MicImagePlugin',
+}
+a.binaries = [
+    (name, path, typ)
+    for name, path, typ in a.binaries
+    if not any(ex in name for ex in _PIL_EXCLUDE_PYDS)
+]
+
 icon_path = os.path.join(os.path.dirname(SPEC), os.pardir, 'icon.ico')
+version_path = os.path.join(os.path.dirname(SPEC), 'version_info.txt')
 if os.path.exists(icon_path):
     a.datas += [('icon.ico', icon_path, 'DATA')]
 
@@ -50,7 +65,7 @@ exe = EXE(
     debug=False,
     bootloader_ignore_signals=False,
     strip=False,
-    upx=True,
+    upx=False,
     upx_exclude=[],
     runtime_tmpdir=None,
     console=False,
@@ -60,4 +75,5 @@ exe = EXE(
     codesign_identity=None,
     entitlements_file=None,
     icon=icon_path if os.path.exists(icon_path) else None,
+    version=version_path if os.path.exists(version_path) else None,
 )

proxy/__init__.py

@@ -1 +1,6 @@
-__version__ = "1.4.0"
+from .config import parse_dc_ip_list, proxy_config
+from .utils import get_link_host
+
+__version__ = "1.6.0"
+
+__all__ = ["__version__", "get_link_host", "proxy_config", "parse_dc_ip_list"]

proxy/bridge.py

@@ -0,0 +1,361 @@
+import asyncio
+import logging
+import struct
+
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from typing import Dict, List, Optional
+
+from .utils import *
+from .stats import stats
+from .config import proxy_config
+from .raw_websocket import RawWebSocket
+
+
+log = logging.getLogger('tg-mtproto-proxy')
+_st_I_le = struct.Struct('<I')
+
+ZERO_64 = b'\x00' * 64
+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'
+}
+
+
+class CryptoCtx:
+    __slots__ = ('clt_dec', 'clt_enc', 'tg_enc', 'tg_dec')
+
+    def __init__(self, clt_dec, clt_enc, tg_enc, tg_dec):
+        self.clt_dec = clt_dec  # decrypt from client
+        self.clt_enc = clt_enc  # encrypt to client
+        self.tg_enc = tg_enc    # encrypt to telegram
+        self.tg_dec = tg_dec    # decrypt from telegram
+
+
+class MsgSplitter:
+    """
+    Splits TCP stream data into individual MTProto transport packets
+    so each can be sent as a separate WS frame.
+    """
+    __slots__ = ('_dec', '_proto', '_cipher_buf', '_plain_buf', '_disabled')
+
+    def __init__(self, relay_init: bytes, proto_int: int):
+        cipher = Cipher(algorithms.AES(relay_init[8:40]),
+                        modes.CTR(relay_init[40:56]))
+        self._dec = cipher.encryptor()
+        self._dec.update(ZERO_64)
+        self._proto = proto_int
+        self._cipher_buf = bytearray()
+        self._plain_buf = bytearray()
+        self._disabled = False
+
+    def split(self, chunk: bytes) -> 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
+
+
+
+async def do_fallback(reader, writer, relay_init, label,
+                       dc, is_media, media_tag,
+                       ctx: CryptoCtx, splitter=None):
+    fallback_dst = DC_DEFAULT_IPS.get(dc)
+    use_cf = proxy_config.fallback_cfproxy
+    cf_first = proxy_config.fallback_cfproxy_priority
+
+    methods: List[str] = ['tcp']
+
+    if use_cf:
+        methods.insert(0 if cf_first else 1, 'cf')
+
+    for method in methods:
+        if method == 'cf':
+            ok = await _cfproxy_fallback(
+                reader, writer, relay_init, label,
+                dc=dc, is_media=is_media,
+                ctx=ctx, splitter=splitter)
+            if ok:
+                return True
+        elif method == 'tcp' and fallback_dst:
+            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, ctx=ctx)
+            if ok:
+                return True
+    return False
+
+
+async def _cfproxy_fallback(reader, writer, relay_init, label,
+                            dc=None, is_media=False,
+                            ctx: CryptoCtx = None, splitter=None):
+    media_tag = ' media' if is_media else ''
+
+    active = proxy_config.active_cfproxy_domain
+    others = [d for d in proxy_config.cfproxy_domains if d != active]
+
+    ws = None
+    chosen_domain = None
+
+    log.info("[%s] DC%d%s -> trying CF proxy",
+            label, dc, media_tag)
+
+    for base_domain in ([active] + others):
+        domain = f'kws{dc}.{base_domain}'
+        try:
+            ws = await RawWebSocket.connect(domain, domain, timeout=10.0)
+            chosen_domain = base_domain
+            break
+        except Exception as exc:
+            log.warning("[%s] DC%d%s CF proxy failed: %s",
+                        label, dc, media_tag, exc)
+
+    if ws is None:
+        return False
+
+    if chosen_domain and chosen_domain != proxy_config.active_cfproxy_domain:
+        log.info("[%s] Switching active CF domain", label)
+        proxy_config.active_cfproxy_domain = chosen_domain
+
+    stats.connections_cfproxy += 1
+    await ws.send(relay_init)
+    await bridge_ws_reencrypt(reader, writer, ws, label,
+                               dc=dc, is_media=is_media,
+                               ctx=ctx, splitter=splitter)
+    return True
+
+
+async def _tcp_fallback(reader, writer, dst, port, relay_init, label,
+                        dc=None, is_media=False, ctx: CryptoCtx = 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, ctx=ctx)
+    return True
+
+
+async def bridge_ws_reencrypt(reader, writer, ws: RawWebSocket, label,
+                               dc=None, is_media=False,
+                               ctx: CryptoCtx = 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_running_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 = ctx.clt_dec.update(chunk)
+                chunk = ctx.tg_enc.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 = ctx.tg_dec.update(data)
+                data = ctx.clt_enc.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_running_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,
+                                ctx: CryptoCtx = 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 = ctx.clt_dec.update(data)
+                    data = ctx.tg_enc.update(plain)
+                else:
+                    stats.bytes_down += n
+                    plain = ctx.tg_dec.update(data)
+                    data = ctx.clt_enc.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

proxy/config.py

@@ -0,0 +1,109 @@
+import logging
+import os
+import string
+import random
+import socket as _socket
+import threading
+
+from dataclasses import dataclass, field
+from typing import Dict, List
+from urllib.request import Request, urlopen
+
+log = logging.getLogger('tg-mtproto-proxy')
+
+CFPROXY_DOMAINS_URL = (
+    "https://raw.githubusercontent.com/Flowseal/tg-ws-proxy/main"
+    "/.github/cfproxy-domains.txt"
+)
+
+_CFPROXY_ENC: List[str] = ['virkgj.com']
+_S = ''.join(chr(c) for c in (46, 99, 111, 46, 117, 107))
+
+
+def _dd(s: str) -> str:
+    """Only for decoding CF proxy domains"""
+    if not s[-4:] == '.com':
+        return s
+    p, n = s[:-4], sum(c.isalpha() for c in s[:-4])
+    return ''.join(
+        chr((ord(c) - (97 if c > '`' else 65) - n) % 26 + (97 if c > '`' else 65))
+        if c.isalpha() else c for c in p
+    ) + _S
+
+
+CFPROXY_DEFAULT_DOMAINS: List[str] = [_dd(d) for d in _CFPROXY_ENC]
+
+
+@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'})
+    buffer_size: int = 256 * 1024
+    pool_size: int = 4
+    fallback_cfproxy: bool = True
+    fallback_cfproxy_priority: bool = True
+    cfproxy_user_domain: str = ''
+    cfproxy_domains: List[str] = field(default_factory=lambda: list(CFPROXY_DEFAULT_DOMAINS))
+    active_cfproxy_domain: str = field(default_factory=lambda: random.choice(CFPROXY_DEFAULT_DOMAINS))
+
+
+proxy_config = ProxyConfig()
+
+
+def _fetch_cfproxy_domain_list() -> List[str]:
+    try:
+        req = Request(CFPROXY_DOMAINS_URL + "?" + "".join(random.choices(string.ascii_letters, k=7)),
+                       headers={'User-Agent': 'tg-ws-proxy'})
+        with urlopen(req, timeout=10) as resp:
+            text = resp.read().decode('utf-8', errors='replace')
+        encoded = [
+            line.strip() for line in text.splitlines()
+            if line.strip() and not line.startswith('#')
+        ]
+        return [_dd(d) for d in encoded]
+    except Exception as exc:
+        log.warning("Failed to fetch CF proxy domain list: %s", exc)
+        return []
+
+
+def refresh_cfproxy_domains() -> None:
+    if proxy_config.cfproxy_user_domain:
+        return
+    
+    fetched = _fetch_cfproxy_domain_list()
+
+    if fetched:
+        seen = set()
+        pool = [d for d in fetched if not (d in seen or seen.add(d))]
+        log.info("CF proxy domain pool updated from GitHub (%d domains)", len(pool))
+    else:
+        pool = list(proxy_config.cfproxy_domains) or list(CFPROXY_DEFAULT_DOMAINS)
+
+    proxy_config.cfproxy_domains = pool
+    proxy_config.active_cfproxy_domain = random.choice(pool)
+
+
+def start_cfproxy_domain_refresh() -> None:
+    threading.Thread(
+        target=refresh_cfproxy_domains,
+        daemon=True,
+        name='cfproxy-domains-refresh',
+    ).start()
+
+
+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

proxy/raw_websocket.py

@@ -0,0 +1,239 @@
+import os
+import ssl
+import base64
+import struct
+import asyncio
+import socket as _socket
+
+from typing import List, Optional, Tuple
+from .config import proxy_config
+
+
+_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')
+
+_ssl_ctx = ssl.create_default_context()
+_ssl_ctx.check_hostname = False
+_ssl_ctx.verify_mode = ssl.CERT_NONE
+
+
+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)
+
+
+def _xor_mask(data: bytes, mask: bytes) -> 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 set_sock_opts(transport, buffer_size):
+    sock = transport.get_extra_info('socket')
+    if sock is None:
+        return
+    
+    try:
+        sock.setsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1)
+    except (OSError, AttributeError):
+        pass
+    
+    try:
+        sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_RCVBUF, buffer_size)
+        sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_SNDBUF, buffer_size)
+    except OSError:
+        pass
+
+
+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(host: str, domain: str, timeout: float = 10.0) -> 'RawWebSocket':
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(host, 443, ssl=_ssl_ctx,
+                                    server_hostname=domain),
+            timeout=min(timeout, 10))
+        
+        set_sock_opts(writer.transport, proxy_config.buffer_size)
+
+        ws_key = base64.b64encode(os.urandom(16)).decode()
+
+        req = (
+            f'GET /apiws 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'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

proxy/stats.py

@@ -0,0 +1,33 @@
+from .utils import human_bytes
+
+class _Stats:
+    def __init__(self):
+        self.connections_total = 0
+        self.connections_active = 0
+        self.connections_ws = 0
+        self.connections_tcp_fallback = 0
+        self.connections_cfproxy = 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} "
+                f"active={self.connections_active} "
+                f"ws={self.connections_ws} "
+                f"tcp_fb={self.connections_tcp_fallback} "
+                f"cf={self.connections_cfproxy} "
+                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()

proxy/tg_ws_proxy.py

@@ -1,11 +1,10 @@
 from __future__ import annotations
 
 import os
-import ssl
 import sys
 import time
-import base64
 import struct
+import random
 import asyncio
 import hashlib
 import argparse
@@ -14,315 +13,30 @@ 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
 
+if __name__ == '__main__' and (__package__ is None or __package__ == ''):
+    _repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    if _repo_root not in sys.path:
+        sys.path.insert(0, _repo_root)
+    __package__ = 'proxy'
 
-@dataclass
+from .utils import *
-class ProxyConfig:
+from .stats import stats
-    port: int = 1443
+from .config import proxy_config, parse_dc_ip_list, start_cfproxy_domain_refresh, CFPROXY_DEFAULT_DOMAINS
-    host: str = '127.0.0.1'
+from .bridge import MsgSplitter, CryptoCtx, do_fallback, bridge_ws_reencrypt
-    secret: str = field(default_factory=lambda: os.urandom(16).hex())
+from .raw_websocket import RawWebSocket, WsHandshakeError, set_sock_opts
-    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('<I')
-
-ZERO_64 = b'\x00' * 64
-
-_ssl_ctx = ssl.create_default_context()
-_ssl_ctx.check_hostname = False
-_ssl_ctx.verify_mode = ssl.CERT_NONE
-
-
-def _set_sock_opts(transport):
-    sock = transport.get_extra_info('socket')
-    if sock is None:
-        return
-    
-    try:
-        sock.setsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1)
-    except (OSError, AttributeError):
-        pass
-    
-    try:
-        sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_RCVBUF, proxy_config.buffer_size)
-        sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_SNDBUF, proxy_config.buffer_size)
-    except OSError:
-        pass
-
-
-def _xor_mask(data: bytes, mask: bytes) -> 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]
@@ -385,132 +99,14 @@ def _generate_relay_init(proto_tag: bytes, dc_idx: int) -> bytes:
     return bytes(result)
 
 
-class _MsgSplitter:
-    """
-    Splits TCP stream data into individual MTProto transport packets
-    so each can be sent as a separate WS frame.
-    """
-    __slots__ = ('_dec', '_proto', '_cipher_buf', '_plain_buf', '_disabled')
-
-    def __init__(self, relay_init: bytes, proto_int: int):
-        cipher = Cipher(algorithms.AES(relay_init[8:40]),
-                        modes.CTR(relay_init[40:56]))
-        self._dec = cipher.encryptor()
-        self._dec.update(ZERO_64)
-        self._proto = proto_int
-        self._cipher_buf = bytearray()
-        self._plain_buf = bytearray()
-        self._disabled = False
-
-    def split(self, chunk: bytes) -> 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 dc == 203:
+        dc = 2
     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_active = 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} "
-                f"active={self.connections_active} "
-                f"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
     
@@ -535,13 +131,13 @@ class _WsPool:
                     or ws.writer.transport.is_closing()):
                 asyncio.create_task(self._quiet_close(ws))
                 continue
-            _stats.pool_hits += 1
+            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
+        stats.pool_misses += 1
         self._schedule_refill(key, target_ip, domains)
         return None
 
@@ -603,194 +199,20 @@ class _WsPool:
                 self._schedule_refill((dc, is_media), target_ip, domains)
         log.info("WS pool warmup started for %d DC(s)", len(dc_redirects))
 
+    def reset(self):
+        self._idle.clear()
+        self._refilling.clear()
+
 _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_running_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_running_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
+    stats.connections_total += 1
-    _stats.connections_active += 1
+    stats.connections_active += 1
     peer = writer.get_extra_info('peername')
     label = f"{peer[0]}:{peer[1]}" if peer else "?"
 
-    _set_sock_opts(writer.transport)
+    set_sock_opts(writer.transport, proxy_config.buffer_size)
 
     try:
         try:
@@ -802,7 +224,7 @@ async def _handle_client(reader, writer, secret: bytes):
 
         result = _try_handshake(handshake, secret)
         if result is None:
-            _stats.connections_bad += 1
+            stats.connections_bad += 1
             log.debug("[%s] bad handshake (wrong secret or proto)", label)
             try:
                 while await reader.read(4096):
@@ -867,27 +289,29 @@ async def _handle_client(reader, writer, secret: bytes):
         
         tg_encryptor.update(ZERO_64)
 
-        dc_key = (dc, is_media)
+        ctx = CryptoCtx(clt_decryptor, clt_encryptor, tg_encryptor, tg_decryptor)
+
+        dc_key = f'{dc}{"m" if is_media else ""}'
         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:
             if dc not in proxy_config.dc_redirects:
-                    log.info("[%s] DC%d not in config -> TCP fallback %s:443",
+                log.info("[%s] DC%d not in config -> fallback",
-                             label, dc, fallback_dst)
+                         label, dc)
-                else:
-                    log.info("[%s] DC%d%s WS blacklisted -> TCP fallback %s:443",
-                             label, dc, media_tag, 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.info("[%s] DC%d%s WS blacklisted -> fallback",
+                         label, dc, media_tag)
+            splitter = None
+            try:
+                splitter = MsgSplitter(relay_init, proto_int)
+            except Exception:
+                pass
+            ok = await do_fallback(
+                reader, writer, relay_init, label,
+                dc, is_media, media_tag,
+                ctx, splitter=splitter)
+            if not ok:
                 log.warning("[%s] DC%d%s no fallback available",
                             label, dc, media_tag)
             return
@@ -917,7 +341,7 @@ async def _handle_client(reader, writer, secret: bytes):
                     all_redirects = False
                     break
                 except WsHandshakeError as exc:
-                    _stats.ws_errors += 1
+                    stats.ws_errors += 1
                     if exc.is_redirect:
                         ws_failed_redirect = True
                         log.warning("[%s] DC%d%s got %d from %s -> %s",
@@ -930,7 +354,7 @@ async def _handle_client(reader, writer, secret: bytes):
                         log.warning("[%s] DC%d%s WS handshake: %s",
                                     label, dc, media_tag, exc.status_line)
                 except Exception as exc:
-                    _stats.ws_errors += 1
+                    stats.ws_errors += 1
                     all_redirects = False
                     log.warning("[%s] DC%d%s WS connect failed: %s",
                                 label, dc, media_tag, exc)
@@ -948,27 +372,26 @@ async def _handle_client(reader, writer, secret: bytes):
                 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
+            splitter_fb = None
-            log.info("[%s] DC%d%s -> TCP fallback to %s:443",
+            try:
-                     label, dc, media_tag, fallback_dst)
+                splitter_fb = MsgSplitter(relay_init, proto_int)
-            ok = await _tcp_fallback(reader, writer, fallback_dst, 443,
+            except Exception:
-                                     relay_init, label, dc=dc,
+                pass
-                                     is_media=is_media,
+            ok = await do_fallback(
-                                     clt_decryptor=clt_decryptor,
+                reader, writer, relay_init, label,
-                                     clt_encryptor=clt_encryptor,
+                dc, is_media, media_tag,
-                                     tg_encryptor=tg_encryptor,
+                ctx, splitter=splitter_fb)
-                                     tg_decryptor=tg_decryptor)
             if ok:
-                log.info("[%s] DC%d%s TCP fallback closed",
+                log.info("[%s] DC%d%s fallback closed",
                          label, dc, media_tag)
             return
 
         dc_fail_until.pop(dc_key, None)
-        _stats.connections_ws += 1
+        stats.connections_ws += 1
 
         splitter = None
         try:
-            splitter = _MsgSplitter(relay_init, proto_int)
+            splitter = MsgSplitter(relay_init, proto_int)
             log.debug("[%s] MsgSplitter activated for proto 0x%08X",
                       label, proto_int)
         except Exception:
@@ -976,13 +399,9 @@ async def _handle_client(reader, writer, secret: bytes):
 
         await ws.send(relay_init)
 
-        await _bridge_ws_reencrypt(reader, writer, ws, label,
+        await bridge_ws_reencrypt(reader, writer, ws, label,
                                    dc=dc, is_media=is_media,
-                                   clt_decryptor=clt_decryptor,
+                                   ctx=ctx, splitter=splitter)
-                                   clt_encryptor=clt_encryptor,
-                                   tg_encryptor=tg_encryptor,
-                                   tg_decryptor=tg_decryptor,
-                                   splitter=splitter)
 
     except asyncio.TimeoutError:
         log.warning("[%s] timeout during handshake", label)
@@ -1000,7 +419,7 @@ async def _handle_client(reader, writer, secret: bytes):
     except Exception as exc:
         log.error("[%s] unexpected: %s", label, exc, exc_info=True)
     finally:
-        _stats.connections_active -= 1
+        stats.connections_active -= 1
         try:
             writer.close()
         except BaseException:
@@ -1009,16 +428,34 @@ async def _handle_client(reader, writer, secret: bytes):
 
 _server_instance = None
 _server_stop_event = None
+_client_tasks: Set[asyncio.Task] = set()
 
 
 async def _run(stop_event: Optional[asyncio.Event] = None):
     global _server_instance, _server_stop_event
     _server_stop_event = stop_event
 
+    _ws_pool.reset()
+    ws_blacklist.clear()
+    dc_fail_until.clear()
+    _client_tasks.clear()
+
+    if proxy_config.fallback_cfproxy:
+        user = proxy_config.cfproxy_user_domain
+        if user:
+            proxy_config.cfproxy_domains = [user]
+            proxy_config.active_cfproxy_domain = user
+        else:
+            proxy_config.cfproxy_domains = list(CFPROXY_DEFAULT_DOMAINS)
+            proxy_config.active_cfproxy_domain = random.choice(CFPROXY_DEFAULT_DOMAINS)
+            start_cfproxy_domain_refresh()
+
     secret_bytes = bytes.fromhex(proxy_config.secret)
 
     def client_cb(r, w):
-        asyncio.create_task(_handle_client(r, w, secret_bytes))
+        task = asyncio.create_task(_handle_client(r, w, secret_bytes))
+        _client_tasks.add(task)
+        task.add_done_callback(_client_tasks.discard)
 
     server = await asyncio.start_server(client_cb, proxy_config.host, proxy_config.port)
     _server_instance = server
@@ -1040,6 +477,10 @@ async def _run(stop_event: Optional[asyncio.Event] = None):
     for dc in sorted(proxy_config.dc_redirects.keys()):
         ip = proxy_config.dc_redirects.get(dc)
         log.info("    DC%d: %s", dc, ip)
+    if proxy_config.fallback_cfproxy:
+        prio = 'CF first' if proxy_config.fallback_cfproxy_priority else 'TCP first'
+        user_domain = "user" if proxy_config.cfproxy_user_domain else "auto"
+        log.info("  CF proxy:      enabled (%s | %s)", prio, user_domain)
     log.info("=" * 60)
     log.info("  Connect link:")
     log.info("    %s", tg_link)
@@ -1049,10 +490,8 @@ async def _run(stop_event: Optional[asyncio.Event] = None):
         try:
             while True:
                 await asyncio.sleep(60)
-                bl = ', '.join(
+                bl = ', '.join(f'DC{k}' for k in sorted(ws_blacklist)) or 'none'
-                    f'DC{d}{"m" if m else ""}'
+                log.info("stats: %s | ws_bl: %s", stats.summary(), bl)
-                    for d, m in sorted(ws_blacklist)) or 'none'
-                log.info("stats: %s | ws_bl: %s", _stats.summary(), bl)
         except asyncio.CancelledError:
             raise
 
@@ -1095,22 +534,6 @@ async def _run(stop_event: Optional[asyncio.Event] = None):
     _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,))
 
@@ -1119,7 +542,7 @@ 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)')
+                    help='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,
@@ -1139,6 +562,13 @@ def main():
                     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)')
+    ap.add_argument('--cfproxy-domain', type=str, default='',
+                    metavar='DOMAIN',
+                    help='User defined Cloudflare-proxied domain for WS fallback')
+    ap.add_argument('--no-cfproxy', action='store_true',
+                    help='Disable Cloudflare proxy fallback')
+    ap.add_argument('--cfproxy-priority', type=bool, default=True,
+                    help='Try cfproxy before tcp fallback (default: true)')
     args = ap.parse_args()
 
     if not args.dc_ip:
@@ -1164,15 +594,15 @@ def main():
         secret_hex = os.urandom(16).hex()
         log.info("Generated secret: %s", secret_hex)
 
-    global proxy_config
+    proxy_config.port = args.port
-    proxy_config = ProxyConfig(
+    proxy_config.host = args.host
-        port=args.port,
+    proxy_config.secret = secret_hex
-        host=args.host,
+    proxy_config.dc_redirects = dc_redirects
-        secret=secret_hex,
+    proxy_config.buffer_size = max(4, args.buf_kb) * 1024
-        dc_redirects=dc_redirects,
+    proxy_config.pool_size = max(0, args.pool_size)
-        buffer_size=max(4, args.buf_kb) * 1024,
+    proxy_config.fallback_cfproxy = not args.no_cfproxy
-        pool_size=max(0, args.pool_size)
+    proxy_config.fallback_cfproxy_priority = args.cfproxy_priority
-    )
+    proxy_config.cfproxy_user_domain = args.cfproxy_domain
 
     log_level = logging.DEBUG if args.verbose else logging.INFO
     log_fmt = logging.Formatter('%(asctime)s  %(levelname)-5s  %(message)s',
@@ -1197,7 +627,7 @@ def main():
     try:
         asyncio.run(_run())
     except KeyboardInterrupt:
-        log.info("Shutting down. Final stats: %s", _stats.summary())
+        log.info("Shutting down. Final stats: %s", stats.summary())
 
 
 if __name__ == '__main__':

proxy/utils.py

@@ -0,0 +1,48 @@
+import socket as _socket
+
+from typing import Optional
+
+
+ZERO_64 = b'\x00' * 64
+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'
+
+
+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 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

pyproject.toml

@@ -7,7 +7,7 @@ name = "tg-ws-proxy"
 dynamic=["version"]
 
 description = "Telegram Desktop WebSocket Bridge Proxy"
-readme = "README.md"
+readme = "docs/README.md"
 requires-python = ">=3.8"
 
 license = { name = "MIT", file = "LICENSE" }
@@ -71,3 +71,6 @@ packages = ["proxy", "ui", "utils"]
 
 [tool.hatch.version]
 path = "proxy/__init__.py"
+
+[tool.ruff.lint]
+ignore = ["F403", "F405"]

ui/ctk_theme.py

@@ -51,8 +51,11 @@ def ctk_theme_for_platform() -> CtkTheme:
     return CtkTheme()
 
 
-def apply_ctk_appearance(ctk: Any) -> None:
+_APPEARANCE_MODE_MAP = {"auto": "system", "light": "Light", "dark": "Dark"}
-    ctk.set_appearance_mode("auto")
+
+
+def apply_ctk_appearance(ctk: Any, mode: str = "auto") -> None:
+    ctk.set_appearance_mode(_APPEARANCE_MODE_MAP.get(mode, "system"))
     ctk.set_default_color_theme("blue")
 
 def center_ctk_geometry(root: Any, width: int, height: int) -> None:

ui/ctk_tray_ui.py

@@ -5,10 +5,11 @@ import webbrowser
 from dataclasses import dataclass
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 
-import proxy.tg_ws_proxy as tg_ws_proxy
+from proxy import __version__, get_link_host, parse_dc_ip_list
-from proxy import __version__
+from proxy.config import CFPROXY_DEFAULT_DOMAINS
 from utils.update_check import RELEASES_PAGE_URL, get_status
 
+
 from ui.ctk_theme import (
     FIRST_RUN_FRAME_PAD,
     CtkTheme,
@@ -27,8 +28,9 @@ _TIP_PORT = (
 _TIP_SECRET = "Секретный ключ для авторизации клиентов"
 _TIP_DC = (
     "Соответствие номера датацентра Telegram (DC) и IP-адреса сервера.\n"
-    "Каждая строка: «номер:IP», например 2:149.154.167.220. "
+    "Каждая строка: «номер:IP», например 4:149.154.167.220. "
-    "Прокси по этим правилам направляет трафик к нужным серверам Telegram"
+    "Прокси по этим правилам направляет трафик к нужным серверам Telegram\n\n"
+    "Если у вас не работают медиа и работает CF-прокси, то попробуйте убрать строку 2:149.154.167.220"
 )
 _TIP_VERBOSE = (
     "Если включено, в файл логов пишется больше подробностей — "
@@ -50,11 +52,143 @@ _TIP_AUTOSTART = (
     "Если вы переместите программу в другую папку, автозапуск сбросится"
 )
 _TIP_CHECK_UPDATES = "При запуске проверять наличие обновлений"
+_TIP_CFPROXY = (
+    "Использовать Cloudflare прокси для недоступных датацентров"
+)
+_TIP_CFPROXY_PRIORITY = (
+    "Пробовать CF-прокси раньше прямого TCP-подключения"
+)
+_TIP_CFPROXY_DOMAIN = (
+    "Ваш собственный домен, проксируемый через Cloudflare, для WS-подключения.\n"
+    "Если не указан — выбирается автоматически из поддерживаемых доменов"
+)
+_TIP_CFPROXY_USER_DOMAIN_CB = (
+    "Указать свой домен вместо автоматического выбора"
+)
 _TIP_SAVE = "Сохранить настройки"
 _TIP_CANCEL = "Закрыть окно без сохранения изменений"
 
+_CFPROXY_HELP_URL = "https://github.com/Flowseal/tg-ws-proxy/blob/main/docs/CfProxy.md"
+_CFPROXY_TEST_DCS = [1, 2, 3, 4, 5, 203]
+
+
+def _run_cfproxy_connectivity_test(domain: str) -> dict:
+    import base64
+    import ssl
+    import socket as _socket
+
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    results = {}
+    for dc in _CFPROXY_TEST_DCS:
+        host = f"kws{dc}.{domain}"
+        try:
+            with _socket.create_connection((host, 443), timeout=5) as raw:
+                with ctx.wrap_socket(raw, server_hostname=host) as ssock:
+                    ws_key = base64.b64encode(os.urandom(16)).decode()
+                    req = (
+                        f"GET /apiws HTTP/1.1\r\n"
+                        f"Host: {host}\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"\r\n"
+                    ).encode()
+                    ssock.sendall(req)
+                    ssock.settimeout(5)
+                    buf = b""
+                    while b"\r\n\r\n" not in buf:
+                        chunk = ssock.recv(512)
+                        if not chunk:
+                            break
+                        buf += chunk
+                    first = buf.decode("utf-8", errors="replace").split("\r\n")[0]
+                    if "101" in first:
+                        results[dc] = True
+                    else:
+                        results[dc] = first or "нет ответа"
+                    ssock.close()
+                raw.close()
+        except _socket.timeout:
+            results[dc] = "таймаут"
+        except OSError as exc:
+            msg = str(exc)
+            results[dc] = msg[:60] if len(msg) > 60 else msg
+    return results
+
+
+def _run_cfproxy_auto_test(domains: list) -> tuple:
+    last: dict = {}
+    for domain in domains:
+        res = _run_cfproxy_connectivity_test(domain)
+        last = res
+        if any(v is True for v in res.values()):
+            return domain, res
+    return None, last
+
+
+def _cfproxy_show_test_results(domain: str, results: dict) -> None:
+    import tkinter as _tk
+    from tkinter import messagebox as _mb
+
+    ok = [dc for dc, v in results.items() if v is True]
+    fail = [(dc, v) for dc, v in results.items() if v is not True]
+    if len(ok) == len(_CFPROXY_TEST_DCS):
+        title = "CF-прокси: всё работает"
+        msg = f"\u2713 Все {len(_CFPROXY_TEST_DCS)} серверов доступны через {domain}."
+    elif not ok:
+        title = "CF-прокси: недоступен"
+        msg = f"\u2717 Ни один сервер не отвечает через {domain}.\n\nОшибки:\n"
+        msg += "\n".join(f"  kws{dc}: {v}" for dc, v in fail)
+    else:
+        title = "CF-прокси: частично работает"
+        msg = (
+            f"Домен: {domain}\n\n"
+            f"\u2713 Работают: {', '.join(f'kws{dc}' for dc in ok)}\n\n"
+            f"\u2717 Недоступны:\n"
+            + "\n".join(f"  kws{dc}: {v}" for dc, v in fail)
+        )
+    root = _tk.Tk()
+    root.withdraw()
+    try:
+        root.attributes("-topmost", True)
+    except Exception:
+        pass
+    _mb.showinfo(title, msg, parent=root)
+    root.destroy()
+
+
+def _cfproxy_show_auto_test_results(ok_domain, results: dict) -> None:
+    import tkinter as _tk
+    from tkinter import messagebox as _mb
+
+    if ok_domain is not None:
+        title = "CF-прокси: доступен"
+        ok = [dc for dc, v in results.items() if v is True]
+        msg = f"\u2713 CF-прокси работает. {len(ok)} из {len(_CFPROXY_TEST_DCS)} серверов доступны."
+    else:
+        title = "CF-прокси: недоступен"
+        msg = "\u2717 Ни один из автоматических CF-доменов не отвечает.\n"
+        msg += "Возможно, блокировка или проблемы с сетью."
+    root = _tk.Tk()
+    root.withdraw()
+    try:
+        root.attributes("-topmost", True)
+    except Exception:
+        pass
+    _mb.showinfo(title, msg, parent=root)
+    root.destroy()
+
 _INNER_W = 396
 
+_APPEARANCE_OPTIONS = ["Авто", "Светлая", "Тёмная"]
+_APPEARANCE_FROM_CFG = {"auto": "Авто", "light": "Светлая", "dark": "Тёмная"}
+_APPEARANCE_TO_CFG = {"Авто": "auto", "Светлая": "light", "Тёмная": "dark"}
+_APPEARANCE_TO_CTK = {"auto": "system", "light": "Light", "dark": "Dark"}
+
 
 def _entry(ctk, parent, theme, *, var=None, width=0, height=36, radius=10, **kw):
     opts = dict(
@@ -155,6 +289,10 @@ class TrayConfigFormWidgets:
     adv_keys: Tuple[str, ...]
     autostart_var: Optional[Any]
     check_updates_var: Optional[Any]
+    cfproxy_var: Optional[Any] = None
+    cfproxy_priority_var: Optional[Any] = None
+    cfproxy_user_domain_var: Optional[Any] = None
+    appearance_var: Optional[Any] = None
 
 
 def install_tray_config_form(
@@ -178,6 +316,33 @@ def install_tray_config_form(
         header, text=f"v{__version__}",
         font=(theme.ui_font_family, 12),
         text_color=theme.text_secondary, anchor="e",
+    ).pack(side="right", padx=(4, 0))
+    appearance_var = ctk.StringVar(
+        value=_APPEARANCE_FROM_CFG.get(cfg.get("appearance", "auto"), "Авто")
+    )
+
+    def _on_appearance_change(choice: str) -> None:
+        cfg_val = _APPEARANCE_TO_CFG.get(choice, "auto")
+        ctk.set_appearance_mode(_APPEARANCE_TO_CTK[cfg_val])
+
+    ctk.CTkComboBox(
+        header,
+        values=_APPEARANCE_OPTIONS,
+        variable=appearance_var,
+        width=102,
+        height=28,
+        font=(theme.ui_font_family, 12),
+        text_color=theme.text_secondary,
+        fg_color=theme.field_bg,
+        border_color=theme.field_border,
+        button_color=theme.field_border,
+        button_hover_color=theme.text_secondary,
+        dropdown_fg_color=theme.field_bg,
+        dropdown_text_color=theme.text_primary,
+        dropdown_hover_color=theme.field_border,
+        corner_radius=8,
+        state="readonly",
+        command=_on_appearance_change,
     ).pack(side="right")
 
     conn = _config_section(ctk, frame, theme, "Подключение MTProto")
@@ -233,6 +398,92 @@ def install_tray_config_form(
     dc_textbox.insert("1.0", "\n".join(cfg.get("dc_ip", default_config["dc_ip"])))
     attach_tooltip_to_widgets([dc_lbl, dc_textbox], _TIP_DC)
 
+    cf_inner = _config_section(ctk, frame, theme, "Cloudflare Proxy")
+
+    cf_row = ctk.CTkFrame(cf_inner, fg_color="transparent")
+    cf_row.pack(fill="x", pady=(0, 4))
+
+    cfproxy_var = ctk.BooleanVar(
+        value=cfg.get("cfproxy", default_config.get("cfproxy", True))
+    )
+    cf_cb = _checkbox(ctk, cf_row, theme, "Включить CF-прокси", cfproxy_var)
+    cf_cb.pack(side="left", padx=(0, 16))
+    attach_ctk_tooltip(cf_cb, _TIP_CFPROXY)
+
+    cfproxy_priority_var = ctk.BooleanVar(
+        value=cfg.get("cfproxy_priority", default_config.get("cfproxy_priority", True))
+    )
+    cf_prio_cb = _checkbox(ctk, cf_row, theme, "Приоритет", cfproxy_priority_var)
+    cf_prio_cb.pack(side="left")
+    attach_ctk_tooltip(cf_prio_cb, _TIP_CFPROXY_PRIORITY)
+
+    _cf_test_btn = [None]
+
+    def _on_cf_test():
+        user_domain = cfproxy_user_domain_var.get().strip() if cf_custom_cb_var.get() else ""
+        btn = _cf_test_btn[0]
+        if btn:
+            btn.configure(text="...", state="disabled")
+        import threading as _threading
+        if user_domain:
+            def _worker():
+                res = _run_cfproxy_connectivity_test(user_domain)
+                if btn:
+                    btn.after(0, lambda: btn.configure(text="Тест", state="normal"))
+                    btn.after(0, lambda: _cfproxy_show_test_results(user_domain, res))
+            _threading.Thread(target=_worker, daemon=True).start()
+        else:
+            def _worker_auto():
+                ok_domain, res = _run_cfproxy_auto_test(CFPROXY_DEFAULT_DOMAINS)
+                if btn:
+                    btn.after(0, lambda: btn.configure(text="Тест", state="normal"))
+                    btn.after(0, lambda: _cfproxy_show_auto_test_results(ok_domain, res))
+            _threading.Thread(target=_worker_auto, daemon=True).start()
+
+    _cf_test_widget = ctk.CTkButton(
+        cf_row, text="Тест", width=56, height=28,
+        font=(theme.ui_font_family, 13), corner_radius=8,
+        fg_color=theme.tg_blue, hover_color=theme.tg_blue_hover,
+        text_color="#ffffff", border_width=1, border_color=theme.field_border,
+        command=_on_cf_test,
+    )
+    _cf_test_widget.pack(side="right")
+    _cf_test_btn[0] = _cf_test_widget
+
+    cf_custom_row = ctk.CTkFrame(cf_inner, fg_color="transparent")
+    cf_custom_row.pack(fill="x")
+
+    saved_user_domain = cfg.get("cfproxy_user_domain", default_config.get("cfproxy_user_domain", ""))
+    cf_custom_cb_var = ctk.BooleanVar(value=bool(saved_user_domain))
+    cf_custom_cb = _checkbox(ctk, cf_custom_row, theme, "Свой домен", cf_custom_cb_var)
+    cf_custom_cb.pack(side="left", padx=(0, 10))
+    attach_ctk_tooltip(cf_custom_cb, _TIP_CFPROXY_USER_DOMAIN_CB)
+
+    ctk.CTkButton(
+        cf_custom_row, text="?", width=28, height=32,
+        font=(theme.ui_font_family, 14), corner_radius=8,
+        fg_color=theme.tg_blue, hover_color=theme.tg_blue_hover,
+        text_color="#ffffff", border_width=1, border_color=theme.field_border,
+        command=lambda: webbrowser.open(_CFPROXY_HELP_URL),
+    ).pack(side="right")
+
+    cfproxy_user_domain_var = ctk.StringVar(value=saved_user_domain)
+    cf_domain_entry = _entry(
+        ctk, cf_custom_row, theme, var=cfproxy_user_domain_var,
+        height=32, radius=8,
+    )
+    cf_domain_entry.pack(side="left", fill="x", expand=True, padx=(0, 6))
+    attach_ctk_tooltip(cf_domain_entry, _TIP_CFPROXY_DOMAIN)
+
+    def _sync_domain_entry(*_):
+        state = "normal" if cf_custom_cb_var.get() else "disabled"
+        cf_domain_entry.configure(state=state)
+        if not cf_custom_cb_var.get():
+            cfproxy_user_domain_var.set("")
+
+    cf_custom_cb_var.trace_add("write", _sync_domain_entry)
+    _sync_domain_entry()
+
     log_inner = _config_section(ctk, frame, theme, "Логи и производительность")
 
     verbose_var = ctk.BooleanVar(value=cfg.get("verbose", False))
@@ -321,6 +572,10 @@ def install_tray_config_form(
         dc_textbox=dc_textbox, verbose_var=verbose_var,
         adv_entries=adv_entries, adv_keys=adv_keys,
         autostart_var=autostart_var, check_updates_var=check_updates_var,
+        cfproxy_var=cfproxy_var,
+        cfproxy_priority_var=cfproxy_priority_var,
+        cfproxy_user_domain_var=cfproxy_user_domain_var,
+        appearance_var=appearance_var,
     )
 
 
@@ -363,12 +618,12 @@ def validate_config_form(
         return "Порт должен быть числом 1-65535"
 
     lines = [
-        l.strip()
+        line.strip()
-        for l in widgets.dc_textbox.get("1.0", "end").strip().splitlines()
+        for line in widgets.dc_textbox.get("1.0", "end").strip().splitlines()
-        if l.strip()
+        if line.strip()
     ]
     try:
-        tg_ws_proxy.parse_dc_ip_list(lines)
+        parse_dc_ip_list(lines)
     except ValueError as e:
         return str(e)
 
@@ -397,6 +652,14 @@ def validate_config_form(
     merge_adv_from_form(widgets, new_cfg, default_config)
     if widgets.check_updates_var is not None:
         new_cfg["check_updates"] = bool(widgets.check_updates_var.get())
+    if widgets.cfproxy_var is not None:
+        new_cfg["cfproxy"] = bool(widgets.cfproxy_var.get())
+    if widgets.cfproxy_priority_var is not None:
+        new_cfg["cfproxy_priority"] = bool(widgets.cfproxy_priority_var.get())
+    if widgets.cfproxy_user_domain_var is not None:
+        new_cfg["cfproxy_user_domain"] = widgets.cfproxy_user_domain_var.get().strip()
+    if widgets.appearance_var is not None:
+        new_cfg["appearance"] = _APPEARANCE_TO_CFG.get(widgets.appearance_var.get(), "auto")
     return new_cfg
 
 
@@ -445,7 +708,7 @@ def populate_first_run_window(
     secret: str,
     on_done: Callable[[bool], None],
 ) -> None:
-    link_host = tg_ws_proxy.get_link_host(host)
+    link_host = get_link_host(host)
     tg_url = f"tg://proxy?server={link_host}&port={port}&secret=dd{secret}"
     fpx, fpy = FIRST_RUN_FRAME_PAD
     frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy)

utils/default_config.py

@@ -17,6 +17,9 @@ _TRAY_DEFAULTS_COMMON: Dict[str, Any] = {
     "log_max_mb": 5,
     "buf_kb": 256,
     "pool_size": 4,
+    "cfproxy": True,
+    "cfproxy_priority": True,
+    "cfproxy_user_domain": "",
 }
 
 

utils/tray_common.py

@@ -14,8 +14,8 @@ from typing import Any, Callable, Dict, Optional, Tuple
 
 import psutil
 
-import proxy.tg_ws_proxy as tg_ws_proxy
+from proxy import __version__, get_link_host, parse_dc_ip_list, proxy_config
-from proxy import __version__
+from proxy.tg_ws_proxy import _run
 from utils.default_config import default_tray_config
 
 log = logging.getLogger("tg-ws-tray")
@@ -60,12 +60,6 @@ def _same_process(meta: dict, proc: psutil.Process, script_hint: str) -> bool:
         return False
     if IS_FROZEN:
         return APP_NAME.lower() in proc.name().lower()
-    try:
-        for arg in proc.cmdline():
-            if script_hint in arg:
-                return True
-    except Exception:
-        pass
     return False
 
 
@@ -76,7 +70,10 @@ def acquire_lock(script_hint: str = "") -> bool:
         try:
             pid = int(f.stem)
         except Exception:
+            try:
                 f.unlink(missing_ok=True)
+            except OSError:
+                pass
             continue
         meta: dict = {}
         try:
@@ -85,12 +82,17 @@ def acquire_lock(script_hint: str = "") -> bool:
                 meta = json.loads(raw)
         except Exception:
             pass
+        is_running = False
         try:
-            if _same_process(meta, psutil.Process(pid), script_hint):
+            is_running = _same_process(meta, psutil.Process(pid), script_hint)
-                return False
         except Exception:
             pass
+        if is_running:
+            return False
+        try:
             f.unlink(missing_ok=True)
+        except OSError:
+            pass
 
     lock_file = APP_DIR / f"{os.getpid()}.lock"
     try:
@@ -100,7 +102,10 @@ def acquire_lock(script_hint: str = "") -> bool:
             encoding="utf-8",
         )
     except Exception:
+        try:
             lock_file.touch()
+        except Exception:
+            pass
     _lock_file_path = lock_file
     return True
 
@@ -234,7 +239,7 @@ def _run_proxy_thread(on_port_busy: Callable[[str], None]) -> None:
     _async_stop = (loop, stop_ev)
 
     try:
-        loop.run_until_complete(tg_ws_proxy._run(stop_event=stop_ev))
+        loop.run_until_complete(_run(stop_event=stop_ev))
     except Exception as exc:
         log.error("Proxy thread crashed: %s", exc)
         if "Address already in use" in str(exc) or "10048" in str(exc):
@@ -252,18 +257,21 @@ def _run_proxy_thread(on_port_busy: Callable[[str], None]) -> None:
 def apply_proxy_config(cfg: dict) -> bool:
     dc_ip_list = cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"])
     try:
-        dc_redirects = tg_ws_proxy.parse_dc_ip_list(dc_ip_list)
+        dc_redirects = parse_dc_ip_list(dc_ip_list)
     except ValueError as e:
         log.error("Bad config dc_ip: %s", e)
         return False
 
-    pc = tg_ws_proxy.proxy_config
+    pc = proxy_config
     pc.port = cfg.get("port", DEFAULT_CONFIG["port"])
     pc.host = cfg.get("host", DEFAULT_CONFIG["host"])
     pc.secret = cfg.get("secret", DEFAULT_CONFIG["secret"])
     pc.dc_redirects = dc_redirects
     pc.buffer_size = max(4, cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"])) * 1024
     pc.pool_size = max(0, cfg.get("pool_size", DEFAULT_CONFIG["pool_size"]))
+    pc.fallback_cfproxy = cfg.get("cfproxy", DEFAULT_CONFIG["cfproxy"])
+    pc.fallback_cfproxy_priority = cfg.get("cfproxy_priority", DEFAULT_CONFIG["cfproxy_priority"])
+    pc.cfproxy_user_domain = cfg.get("cfproxy_user_domain", DEFAULT_CONFIG["cfproxy_user_domain"])
     return True
 
 
@@ -277,7 +285,7 @@ def start_proxy(cfg: dict, on_error: Callable[[str], None]) -> None:
         on_error("Ошибка конфигурации DC → IP.")
         return
 
-    pc = tg_ws_proxy.proxy_config
+    pc = proxy_config
     log.info("Starting proxy on %s:%d ...", pc.host, pc.port)
     _proxy_thread = threading.Thread(
         target=_run_proxy_thread, args=(on_error,), daemon=True, name="proxy"
@@ -307,7 +315,7 @@ def tg_proxy_url(cfg: dict) -> str:
     host = cfg.get("host", DEFAULT_CONFIG["host"])
     port = cfg.get("port", DEFAULT_CONFIG["port"])
     secret = cfg.get("secret", DEFAULT_CONFIG["secret"])
-    link_host = tg_ws_proxy.get_link_host(host)
+    link_host = get_link_host(host)
     return f"tg://proxy?server={link_host}&port={port}&secret=dd{secret}"
 
 
@@ -393,7 +401,7 @@ _ctk_root: Any = None
 _ctk_root_ready = threading.Event()
 
 
-def ensure_ctk_thread(ctk: Any) -> bool:
+def ensure_ctk_thread(ctk: Any, mode: str = "auto") -> bool:
     global _ctk_root
     if ctk is None:
         return False
@@ -405,7 +413,7 @@ def ensure_ctk_thread(ctk: Any) -> bool:
         from ui.ctk_theme import apply_ctk_appearance, install_tkinter_variable_del_guard
 
         install_tkinter_variable_del_guard()
-        apply_ctk_appearance(ctk)
+        apply_ctk_appearance(ctk, mode)
         _ctk_root = ctk.CTk()
         _ctk_root.withdraw()
         _ctk_root_ready.set()

utils/win32_theme.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import sys
+
+def is_windows_dark_theme() -> bool:
+    if sys.platform != "win32":
+        return False
+
+    try:
+        import winreg
+        key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Themes\Personalize")
+        value, _ = winreg.QueryValueEx(key, "AppsUseLightTheme")
+        return value == 0
+    except Exception:
+        return False
+
+def apply_windows_dark_theme() -> None:
+    try:
+        import ctypes
+        uxtheme = ctypes.windll.uxtheme
+        
+        try:
+            set_preferred = uxtheme[135]
+            result = set_preferred(2)
+            if result == 0:
+                flush = uxtheme[136]
+                flush()
+        except Exception:
+            try:
+                allow_dark = uxtheme[135]
+                allow_dark(True)
+            except Exception:
+                pass
+    except Exception:
+        pass

windows.py

@@ -30,8 +30,12 @@ try:
 except ImportError:
     Image = None
 
-import proxy.tg_ws_proxy as tg_ws_proxy
+from proxy import get_link_host
 
+from utils.win32_theme import (
+    is_windows_dark_theme, 
+    apply_windows_dark_theme,
+)
 from utils.tray_common import (
     APP_NAME, DEFAULT_CONFIG, FIRST_RUN_MARKER, IS_FROZEN, LOG_FILE,
     acquire_lock, bootstrap, check_ipv6_warning, ctk_run_dialog,
@@ -196,7 +200,7 @@ def _on_exit(icon=None, item=None) -> None:
 # settings dialog
 
 def _edit_config_dialog() -> None:
-    if not ensure_ctk_thread(ctk):
+    if not ensure_ctk_thread(ctk, _config.get("appearance", "auto")):
         _show_error("customtkinter не установлен.")
         return
 
@@ -262,7 +266,7 @@ def _show_first_run() -> None:
     ensure_dirs()
     if FIRST_RUN_MARKER.exists():
         return
-    if not ensure_ctk_thread(ctk):
+    if not ensure_ctk_thread(ctk, _config.get("appearance", "auto")):
         FIRST_RUN_MARKER.touch()
         return
 
@@ -297,7 +301,7 @@ 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)
+    link_host = get_link_host(host)
     return pystray.Menu(
         pystray.MenuItem(f"Открыть в Telegram ({link_host}:{port})", _on_open_in_telegram, default=True),
         pystray.MenuItem("Скопировать ссылку", _on_copy_link),
@@ -316,6 +320,10 @@ def run_tray() -> None:
     global _tray_icon, _config
 
     _config = load_config()
+
+    if is_windows_dark_theme:
+        apply_windows_dark_theme()
+
     bootstrap(_config)
 
     if pystray is None or Image is None or ctk is None: