macos build fix

.github/github-bot.yml

@@ -0,0 +1,2 @@
+trusted_users:
+  - Flowseal

.github/workflows/build.yml

@@ -12,6 +12,31 @@ on:
         description: "Release version tag (e.g. v1.0.0)"
         required: false
         default: "v1.0.0"
+      build_windows_x64:
+        description: 'Build Windows x64?'
+        type: boolean
+        required: false
+        default: true
+      build_windows_arm64:
+        description: 'Build Windows ARM64?'
+        type: boolean
+        required: false
+        default: true
+      build_win7:
+        description: 'Build Windows 7 (x64 + x86)?'
+        type: boolean
+        required: false
+        default: true
+      build_macos:
+        description: 'Build macOS universal?'
+        type: boolean
+        required: false
+        default: true
+      build_linux:
+        description: 'Build Linux (amd64/.deb/.rpm)?'
+        type: boolean
+        required: false
+        default: true
 
 permissions:
   contents: write
@@ -19,6 +44,7 @@ permissions:
 jobs:
   build-windows-x64:
     runs-on: windows-latest
+    if: ${{ github.event.inputs.build_windows_x64 == 'true' }}
     steps:
       - name: Checkout
         uses: actions/checkout@v6
@@ -78,6 +104,7 @@ jobs:
 
   build-windows-arm64:
     runs-on: windows-11-arm
+    if: ${{ github.event.inputs.build_windows_arm64 == 'true' }}
     env:
       CRYPTOGRAPHY_VERSION: "46.0.5"
       ARM64_WHEELHOUSE: wheelhouse-arm64
@@ -154,6 +181,7 @@ jobs:
 
   build-win7:
     runs-on: windows-latest
+    if: ${{ github.event.inputs.build_win7 == 'true' }}
     strategy:
       matrix:
         include:
@@ -207,6 +235,9 @@ jobs:
 
   build-macos:
     runs-on: macos-latest
+    if: ${{ github.event.inputs.build_macos == 'true' }}
+    env:
+      CFFI_VERSION: "2.0.0"
     steps:
       - name: Checkout
         uses: actions/checkout@v6
@@ -232,7 +263,7 @@ jobs:
             --python-version 3.12 \
             --implementation cp \
             -d wheelhouse/arm64 \
-            'cffi>=2.0.0' \
+            "cffi==$CFFI_VERSION" \
             Pillow==12.1.0 \
             psutil==7.0.0
 
@@ -242,7 +273,7 @@ jobs:
             --python-version 3.12 \
             --implementation cp \
             -d wheelhouse/x86_64 \
-            'cffi>=2.0.0' \
+            "cffi==$CFFI_VERSION" \
             Pillow==12.1.0
 
           python3.12 -m pip download \
@@ -345,6 +376,7 @@ jobs:
 
   build-linux:
     runs-on: ubuntu-latest
+    if: ${{ github.event.inputs.build_linux == 'true' }}
     steps:
       - name: Checkout
         uses: actions/checkout@v6
@@ -511,7 +543,13 @@ jobs:
   release:
     needs: [build-windows-x64, build-windows-arm64, build-win7, build-macos, build-linux]
     runs-on: ubuntu-latest
-    if: ${{ github.event.inputs.make_release == 'true' }}
+    if: >-
+      ${{ github.event.inputs.make_release == 'true'
+          && github.event.inputs.build_windows_x64 == 'true'
+          && github.event.inputs.build_windows_arm64 == 'true'
+          && github.event.inputs.build_win7 == 'true'
+          && github.event.inputs.build_macos == 'true'
+          && github.event.inputs.build_linux == 'true' }}
     steps:
       - uses: actions/download-artifact@v8
         with:
@@ -525,7 +563,7 @@ jobs:
           tag_name: ${{ github.event.inputs.version }}
           name: "TG WS Proxy ${{ github.event.inputs.version }}"
           body: |
-            ## 
+            --- 
             ### [❤️ Поддержать развитие проекта](https://github.com/Flowseal/tg-ws-proxy/blob/main/docs/Funding.md)
 
             > [!TIP]

docs/CfWorker.md

@@ -33,6 +33,7 @@ workers.dev
 
 7. Скопируйте домен из поля справа и укажите его в настройках **Cloudflare Worker** (или через аргумент `--cfproxy-worker-domain`)
     * Пример домена: `random-symbols-1234.username.workers.dev`  
+	* **Можно указывать несколько доменов через запятую (или повторением аргумента `--cfproxy-worker-domain`)**
    <img width="414" height="182" alt="image" src="https://github.com/user-attachments/assets/4fb0b111-8026-4d17-b993-6c70ec37f1f5" />
 
 

docs/README.md

@@ -46,6 +46,7 @@
   - **[Docker](./README.docker.md)**
 - [Настройка Cloudflare Worker'а (бесплатный аналог CF-прокси)](./CfWorker.md)
 - [Настройка Cloudflare-домена (CF-прокси)](./CfProxy.md)
+- [Тестовое окружение Telegram (тестовые DC)](./TestDc.md)
 - [Fake TLS + upstream в Nginx](./FakeTlsNginx.md)
 - [Файлы конфигурации Tray-приложения](./TrayConfig.md)
 - [Установка из исходников](./BuildFromSource.md)

docs/TestDc.md

@@ -0,0 +1,19 @@
+# Тестовое окружение Telegram (тестовые DC)
+
+Маршрутизация трафика к тестовым дата-центрам Telegram (test environment).
+Нужна для разработки/тестирования ботов и клиентов в тестовой среде Telegram.
+
+## Как включается
+
+**Автоматически.** Telegram Desktop помечает тестовые DC сдвигом +10000
+(10001–10003). Прокси распознаёт сдвиг сам — включать ничего не нужно, можно
+одновременно пользоваться продовым и тестовым аккаунтом в одном клиенте.
+
+**Принудительно.** Для клиентов, которые сообщают тестовые DC как обычные 1-3
+(Telethon, TDLib) — распознать их автоматически нельзя. Тогда весь трафик
+принудительно направляется на тестовые DC (продовые аккаунты через этот прокси
+работать перестанут). Для принудительной работы замените в конфиге force_test_dc на true или используйте флаг `--force-test-dc` в CLI
+
+## Ограничения
+
+Работает только для маршрутов **прямой DC → IP** и **Cloudflare Worker** (см. [Настройка Cloudflare Worker'а](./CfWorker.md)).

docs/TrayConfig.md

@@ -23,6 +23,7 @@ Tray-приложение хранит данные в:
   "cfproxy": true,
   "cfproxy_user_domain": "",
   "cfproxy_worker_domain": "",
+  "force_test_dc": false,
   "appearance": "auto"
 }
 ```

linux.py

@@ -188,6 +188,8 @@ def _edit_config_dialog() -> None:
                 messagebox.showerror(t("app.error_title"), merged, parent=root)
                 return
 
+            merged["force_test_dc"] = _config.get("force_test_dc", DEFAULT_CONFIG["force_test_dc"])
+
             _ui_only_keys = {"appearance", "check_updates", "language"}
             config_changed = any(merged.get(k) != _config.get(k) for k in merged)
             proxy_changed = any(merged.get(k) != _config.get(k) for k in merged if k not in _ui_only_keys)

macos.py

@@ -475,6 +475,7 @@ def _edit_config_dialog() -> None:
         "cfproxy": cfproxy,
         "cfproxy_user_domain": cfproxy_domains,
         "cfproxy_worker_domain": cfworker_domains,
+        "force_test_dc": cfg.get("force_test_dc", DEFAULT_CONFIG["force_test_dc"]),
     }
     save_config(new_cfg)
     log.info("Config saved: %s", new_cfg)

packaging/version_info.txt

@@ -4,8 +4,8 @@
 # http://msdn.microsoft.com/en-us/library/ms646997.aspx
 VSVersionInfo(
   ffi=FixedFileInfo(
-    filevers=(1, 8, 0, 0),
+    filevers=(1, 9, 0, 0),
-    prodvers=(1, 8, 0, 0),
+    prodvers=(1, 9, 0, 0),
     mask=0x3f,
     flags=0x0,
     OS=0x40004,
@@ -21,12 +21,12 @@ VSVersionInfo(
           [
             StringStruct(u'CompanyName', u'Flowseal'),
             StringStruct(u'FileDescription', u'Telegram Desktop WebSocket Bridge Proxy'),
-            StringStruct(u'FileVersion', u'1.8.0.0'),
+            StringStruct(u'FileVersion', u'1.9.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.8.0.0'),
+            StringStruct(u'ProductVersion', u'1.9.0.0'),
           ]
         )
       ]

proxy/__init__.py

@@ -1,6 +1,6 @@
 from .config import parse_dc_ip_list, proxy_config, coerce_domain_list
 from .utils import get_link_host, build_github_opener
 
-__version__ = "1.8.0"
+__version__ = "1.9.0"
 
 __all__ = ["__version__", "get_link_host", "proxy_config", "parse_dc_ip_list", "build_github_opener", "coerce_domain_list"]

proxy/bridge.py

@@ -130,10 +130,11 @@ class MsgSplitter:
 
 
 async def do_fallback(reader, writer, relay_init, label,
-                       dc: int, is_media: bool, media_tag: str,
+                       dc: int, is_test_dc: bool, is_media: bool, media_tag: str,
                        ctx: CryptoCtx, splitter=None):
-    fallback_dst = DC_DEFAULT_IPS.get(dc)
+    ip_table = DC_TEST_IPS if is_test_dc else DC_DEFAULT_IPS
-    use_cf = proxy_config.fallback_cfproxy
+    fallback_dst = ip_table.get(dc)
+    use_cf = proxy_config.fallback_cfproxy and not is_test_dc
     worker_domains = proxy_config.cfproxy_worker_domains
 
     methods: List[str] = []
@@ -149,8 +150,8 @@ async def do_fallback(reader, writer, relay_init, label,
         if method == 'cf_worker' and fallback_dst:
             ok = await _cfproxy_worker_fallback(
                 reader, writer, relay_init, label, ctx,
-                dc=dc, is_media=is_media, fallback_dst=fallback_dst,
+                dc=dc, is_test_dc=is_test_dc, is_media=is_media,
-                splitter=splitter)
+                fallback_dst=fallback_dst, splitter=splitter)
             if ok:
                 return True
         elif method == 'cf':
@@ -173,7 +174,7 @@ async def do_fallback(reader, writer, relay_init, label,
 
 async def _cfproxy_worker_fallback(reader, writer, relay_init, label,
                                    ctx: CryptoCtx,
-                                   dc: int, is_media: bool,
+                                   dc: int, is_test_dc: bool, is_media: bool,
                                    fallback_dst: str,
                                    splitter=None):
     media_tag = ' media' if is_media else ''
@@ -184,7 +185,7 @@ async def _cfproxy_worker_fallback(reader, writer, relay_init, label,
     random.shuffle(worker_domains)
 
     for worker_domain in worker_domains:
-        ws = await cf_worker_pool.get(dc, worker_domain, fallback_dst)
+        ws = None if is_test_dc else await cf_worker_pool.get(dc, worker_domain, fallback_dst)
         if ws:
             log.info("[%s] DC%d%s -> CF worker pool hit for %s",
                      label, dc, media_tag, fallback_dst)

proxy/config.py

@@ -72,6 +72,7 @@ class ProxyConfig:
     cfproxy_worker_domains: List[str] = field(default_factory=list)
     fake_tls_domain: str = ''
     proxy_protocol: bool = False
+    force_test_dc: bool = False
 
 
 proxy_config = ProxyConfig()

proxy/pool.py

@@ -15,14 +15,21 @@ log = logging.getLogger('tg-mtproto-proxy')
 
 class _WsPool:
     WS_POOL_MAX_AGE = 120.0
+    WS_POOL_CHECK_INTERVAL = 5.0
+    REFILL_BACKOFF_INITIAL = 60.0
+    REFILL_BACKOFF_MAX = 3600.0
     
     def __init__(self):
         self._idle: Dict[Tuple[int, bool], deque] = {}
         self._refilling: Set[Tuple[int, bool]] = set()
-        self.fronting_until: float = 0.0
+        self._rotating: Dict[Tuple[int, bool], asyncio.Task] = {}
+        self._refill_failures: Dict[Tuple[int, bool], int] = {}
+        self._refill_after: Dict[Tuple[int, bool], float] = {}
+        self.try_fronting_first = False
 
     async def get(self, dc: int, is_media: bool,
-                  target_ip: str, domains: List[str]
+                  target_ip: str, domains: List[str],
+                  *, allow_refill: bool = True
                   ) -> Optional[RawWebSocket]:
         key = (dc, is_media)
         now = time.monotonic()
@@ -41,19 +48,28 @@ class _WsPool:
             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)
+            self.report_success(dc, is_media)
+            if allow_refill:
+                self._schedule_refill(key, target_ip, domains)
             return ws
 
         stats.pool_misses += 1
-        self._schedule_refill(key, target_ip, domains)
+        if allow_refill:
+            self._schedule_refill(key, target_ip, domains)
         return None
 
     def _schedule_refill(self, key, target_ip, domains):
-        if key in self._refilling:
+        if (key in self._refilling
+                or time.monotonic() < self._refill_after.get(key, 0)):
             return
         self._refilling.add(key)
         asyncio.create_task(self._refill(key, target_ip, domains))
 
+    def report_success(self, dc: int, is_media: bool) -> None:
+        key = (dc, is_media)
+        self._refill_failures.pop(key, None)
+        self._refill_after.pop(key, None)
+
     async def _refill(self, key, target_ip, domains):
         dc, is_media = key
         try:
@@ -61,27 +77,98 @@ class _WsPool:
             needed = proxy_config.pool_size - len(bucket)
             if needed <= 0:
                 return
+            connected = 0
             tasks = [asyncio.create_task(
-                self._connect_one(target_ip, domains, time.monotonic() < self.fronting_until))
+                self._connect_one(target_ip, domains))
                 for _ in range(needed)]
             for t in tasks:
                 try:
                     ws = await t
                     if ws:
                         bucket.append((ws, time.monotonic()))
+                        connected += 1
+                        self._schedule_rotation(key, target_ip, domains)
                 except Exception:
                     pass
+            if connected:
+                self.report_success(dc, is_media)
+            else:
+                failures = self._refill_failures.get(key, 0) + 1
+                self._refill_failures[key] = failures
+                delay = min(
+                    self.REFILL_BACKOFF_INITIAL
+                    * (2 ** min(failures - 1, 6)),
+                    self.REFILL_BACKOFF_MAX,
+                )
+                self._refill_after[key] = time.monotonic() + delay
+                log.info(
+                    "WS pool refill failed for DC%d%s, retry in %.0fs",
+                    dc, 'm' if is_media else '', delay)
             log.debug("WS pool refilled DC%d%s: %d ready",
                       dc, 'm' if is_media else '', len(bucket))
         finally:
             self._refilling.discard(key)
 
-    @staticmethod
+    def _schedule_rotation(self, key, target_ip, domains):
-    async def _connect_one(target_ip, domains, fronting_active) -> Optional[RawWebSocket]:
+        if key in self._rotating:
+            return
+        self._rotating[key] = asyncio.create_task(
+            self._rotate(key, target_ip, domains))
+
+    async def _rotate(self, key, target_ip, domains):
+        dc, is_media = key
+        try:
+            while True:
+                bucket = self._idle.get(key)
+                if not bucket:
+                    return
+
+                expires_at = min(
+                    created + self.WS_POOL_MAX_AGE
+                    for _, created in bucket)
+                await asyncio.sleep(min(
+                    self.WS_POOL_CHECK_INTERVAL,
+                    max(0, expires_at - time.monotonic())))
+
+                now = time.monotonic()
+                expired = []
+                ready = deque()
+                while bucket:
+                    ws, created = bucket.popleft()
+                    if (now - created >= self.WS_POOL_MAX_AGE
+                            or ws._closed
+                            or ws.writer.transport.is_closing()):
+                        expired.append(ws)
+                    else:
+                        ready.append((ws, created))
+                bucket.extend(ready)
+
+                if expired:
+                    for ws in expired:
+                        asyncio.create_task(self._quiet_close(ws))
+                    log.debug(
+                        "WS pool rotated DC%d%s: %d stale, %d ready",
+                        dc, 'm' if is_media else '', len(expired), len(bucket))
+                    self._schedule_refill(key, target_ip, domains)
+        finally:
+            if self._rotating.get(key) is asyncio.current_task():
+                self._rotating.pop(key, None)
+
+    async def _connect_one(self, target_ip, domains) -> Optional[RawWebSocket]:
         for domain in domains:
+            if self.try_fronting_first:
+                ws = await self._connect_fronted(target_ip, domain)
+                if ws:
+                    return ws
             try:
-                return await RawWebSocket.connect(
+                ws = await RawWebSocket.connect(
-                    target_ip, domain, timeout=8, sni="sprinthost.ru" if fronting_active else None)
+                    target_ip, domain, timeout=8)
+                self.try_fronting_first = False
+                return ws
+            except asyncio.TimeoutError:
+                if self.try_fronting_first:
+                    return None
+                return await self._connect_fronted(target_ip, domain)
             except WsHandshakeError as exc:
                 if exc.is_redirect:
                     continue
@@ -90,8 +177,18 @@ class _WsPool:
                 return None
         return None
 
-    @staticmethod
+    async def _connect_fronted(self, target_ip, domain) -> Optional[RawWebSocket]:
-    async def _quiet_close(ws):
+        try:
+            ws = await RawWebSocket.connect(
+                target_ip, domain, timeout=7, sni="sprinthost.ru")
+        except Exception:
+            return None
+
+        stats.connections_fronting += 1
+        self.try_fronting_first = True
+        return ws
+
+    async def _quiet_close(self, ws):
         try:
             await ws.close()
         except Exception:
@@ -107,9 +204,16 @@ class _WsPool:
         log.info("WS pool warmup started for %d DC(s)", len(proxy_config.dc_redirects))
 
     def reset(self):
+        loop = asyncio.get_running_loop()
+        for task in self._rotating.values():
+            if not task.done() and task.get_loop() is loop:
+                task.cancel()
         self._idle.clear()
         self._refilling.clear()
-        self.fronting_until = 0.0
+        self._rotating.clear()
+        self._refill_failures.clear()
+        self._refill_after.clear()
+        self.try_fronting_first = False
 
 
 class _CfWorkerPool:
@@ -172,8 +276,7 @@ class _CfWorkerPool:
         finally:
             self._refilling.discard(key)
 
-    @staticmethod
+    async def _connect_one(self, worker_domain, fallback_dst, dc) -> Optional[RawWebSocket]:
-    async def _connect_one(worker_domain, fallback_dst, dc) -> Optional[RawWebSocket]:
         query = urlencode({
             'dst': fallback_dst,
             'dc': str(dc),
@@ -185,8 +288,7 @@ class _CfWorkerPool:
         except Exception:
             return None
 
-    @staticmethod
+    async def _quiet_close(self, ws):
-    async def _quiet_close(ws):
         try:
             await ws.close()
         except Exception:

proxy/tg_ws_proxy.py

@@ -33,14 +33,14 @@ from ._aes import Cipher, algorithms, modes
 
 log = logging.getLogger('tg-mtproto-proxy')
 
-DC_FAIL_COOLDOWN = 30.0
+IP_FAIL_COOLDOWN = 3600.0
+DC_FAIL_COOLDOWN = 60.0
 WS_FAIL_TIMEOUT = 2.0
-FRONTING_COOLDOWN = 1800.0
 LISTENER_CHECK_INTERVAL = 5.0
 LISTENER_RESTART_DELAY = 1.0
 ws_blacklist: Set[str] = set()
 dc_fail_until: Dict[str, float] = {}
-fronting_until: float = 0.0
+ip_fail_until: Dict[str, float] = {}
 
 
 def _try_handshake(handshake: bytes, secret: bytes) -> Optional[Tuple[int, bool, bytes, bytes]]:
@@ -248,8 +248,6 @@ def _build_crypto_ctx(client_dec_prekey_iv, secret, relay_init):
 
 
 async def _handle_client(reader, writer, secret: bytes):
-    global fronting_until
-    
     stats.connections_total += 1
     stats.connections_active += 1
     peer = writer.get_extra_info('peername')
@@ -278,6 +276,11 @@ async def _handle_client(reader, writer, secret: bytes):
 
         dc, is_media, proto_tag, client_dec_prekey_iv = result
 
+        is_test_dc = proxy_config.force_test_dc or dc >= 10000
+        if dc >= 10000:
+            log.info("[%s] test DC%d -> DC%d", label, dc, dc - 10000)
+            dc -= 10000
+
         if proto_tag == PROTO_TAG_ABRIDGED:
             proto_int = PROTO_ABRIDGED_INT
         elif proto_tag == PROTO_TAG_INTERMEDIATE:
@@ -293,17 +296,27 @@ async def _handle_client(reader, writer, secret: bytes):
         relay_init = _generate_relay_init(proto_tag, dc_idx)
         ctx = _build_crypto_ctx(client_dec_prekey_iv, secret, relay_init)
 
-        dc_key = f'{dc}{"m" if is_media else ""}'
+        dc_key = f'{dc}{"t" if is_test_dc else ""}{"m" if is_media else ""}'
         media_tag = " media" if is_media else ""
+        now = time.monotonic()
+        ws_path = WS_PATH_TEST if is_test_dc else WS_PATH
+        target = proxy_config.dc_redirects.get(dc)
+        is_any_cf_fallback = proxy_config.fallback_cfproxy or proxy_config.cfproxy_worker_domains
+
+        # Fallback if DC not in config, if WS blacklisted for this DC/is_media or if connect to ip is timed out
+        if (dc not in proxy_config.dc_redirects
+            or dc_key in ws_blacklist
+            or now < ip_fail_until.get(target, 0) and is_any_cf_fallback):
 
-        # 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:
             if dc not in proxy_config.dc_redirects:
                 log.info("[%s] DC%d not in config -> fallback",
                          label, dc)
-            else:
+            elif dc_key in ws_blacklist:
                 log.info("[%s] DC%d%s WS blacklisted -> fallback",
                          label, dc, media_tag)
+            else:
+                log.info("[%s] DC%d%s WS connect to %s was timed out -> fallback",
+                         label, dc, media_tag, target)
             splitter = None
             try:
                 splitter = MsgSplitter(relay_init, proto_int)
@@ -311,56 +324,38 @@ async def _handle_client(reader, writer, secret: bytes):
                 pass
             ok = await do_fallback(
                 clt_reader, clt_writer, relay_init, label,
-                dc, is_media, media_tag,
+                dc, is_test_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
 
-        now = time.monotonic()
+        ws_timeout = WS_FAIL_TIMEOUT if now < dc_fail_until.get(dc_key, 0) else 5.0
-        fail_until = dc_fail_until.get(dc_key, 0)
-        ws_timeout = WS_FAIL_TIMEOUT if now < fail_until else 10.0
-        fronting_active = now < fronting_until
 
         domains = ws_domains(dc, is_media)
-        target = proxy_config.dc_redirects[dc]
         ws = None
         ws_failed_redirect = False
         ws_timed_out = False
         all_redirects = True
 
-        ws = await ws_pool.get(dc, is_media, target, domains)
+        allow_pool_refill = now >= ip_fail_until.get(target, 0)
+        ws = await ws_pool.get(
+            dc, is_media, target, domains,
+            allow_refill=allow_pool_refill,
+        ) if not is_test_dc else None
         if ws:
             log.info("[%s] DC%d%s -> pool hit via %s",
                      label, dc, media_tag, target)
-        elif fronting_active:
-            # TODO: Move fronting logic into bridge.py where other fallbacks are handled
-            log.info("[%s] DC%d%s -> fronting / Host %s",
-                     label, dc, media_tag, domains[0])
-            try:
-                ws = await RawWebSocket.connect(target, domains[0],
-                                                timeout=10.0,
-                                                sni="sprinthost.ru")
-            except Exception as exc:
-                stats.ws_errors += 1
-                log.warning("[%s] DC%d%s fronting failed: %s",
-                            label, dc, media_tag, repr(exc))
-            if ws:
-                stats.connections_fronting += 1
-                fronting_until = now + FRONTING_COOLDOWN
-                ws_pool.fronting_until = fronting_until
-            else:
-                fronting_until = 0.0
-                ws_pool.fronting_until = 0.0
         else:
             for domain in domains:
-                url = f'wss://{domain}/apiws'
+                url = f'wss://{domain}{ws_path}'
                 log.info("[%s] DC%d%s -> %s via %s",
                          label, dc, media_tag, url, target)
                 try:
                     ws = await RawWebSocket.connect(target, domain,
-                                                    timeout=ws_timeout)
+                                                    timeout=ws_timeout,
+                                                    path=ws_path)
                     all_redirects = False
                     break
                 except WsHandshakeError as exc:
@@ -381,35 +376,20 @@ async def _handle_client(reader, writer, secret: bytes):
                     ws_timed_out = True
                     log.warning("[%s] DC%d%s WS connect timed out via %s",
                                 label, dc, media_tag, domain)
+                    break
                 except Exception as exc:
                     stats.ws_errors += 1
                     all_redirects = False
                     log.warning("[%s] DC%d%s WS connect failed: %s",
                                 label, dc, media_tag, repr(exc))
 
-        # Fronting fallback if WS timed out
-        # TODO: Move fronting logic into bridge.py where other fallbacks are handled
-        # and don't forget about WsPool fronting fallback
-        if ws is None and ws_timed_out and not fronting_active:
-            log.info("[%s] DC%d%s -> fronting fallback (Host %s)",
-                     label, dc, media_tag, domains[0])
-            try:
-                ws = await RawWebSocket.connect(target, domains[0],
-                                                timeout=10.0,
-                                                sni="sprinthost.ru")
-            except Exception as exc:
-                stats.ws_errors += 1
-                log.warning("[%s] DC%d%s fronting failed: %s",
-                            label, dc, media_tag, repr(exc))
-            if ws:
-                fronting_until = now + FRONTING_COOLDOWN
-                ws_pool.fronting_until = now + FRONTING_COOLDOWN
-                stats.connections_fronting += 1
-                log.info("[%s] DC%d%s fronting OK for %ds",
-                         label, dc, media_tag, int(FRONTING_COOLDOWN))
-
         # WS failed -> fallback
         if ws is None:
+            if ws_timed_out:
+                ip_fail_until[target] = now + IP_FAIL_COOLDOWN
+                log.info("[%s] DC%d%s WS connect to %s timed out, cooldown for %ds",
+                         label, dc, media_tag, target, int(IP_FAIL_COOLDOWN))
+
             if ws_failed_redirect and all_redirects:
                 ws_blacklist.add(dc_key)
                 log.warning("[%s] DC%d%s blacklisted for WS (all 302)",
@@ -418,8 +398,6 @@ async def _handle_client(reader, writer, secret: bytes):
                 dc_fail_until[dc_key] = now + DC_FAIL_COOLDOWN
             else:
                 dc_fail_until[dc_key] = now + DC_FAIL_COOLDOWN
-                # TODO: may be don't try regular WS connection and do fallback instanstly
-                # instead of waiting for WS_FAIL_TIMEOUT and then fallback
                 log.info("[%s] DC%d%s WS cooldown for %ds",
                          label, dc, media_tag, int(DC_FAIL_COOLDOWN))
 
@@ -430,7 +408,7 @@ async def _handle_client(reader, writer, secret: bytes):
                 pass
             ok = await do_fallback(
                 clt_reader, clt_writer, relay_init, label,
-                dc, is_media, media_tag,
+                dc, is_test_dc, is_media, media_tag,
                 ctx, splitter=splitter_fb)
             if ok:
                 log.info("[%s] DC%d%s fallback closed",
@@ -438,6 +416,8 @@ async def _handle_client(reader, writer, secret: bytes):
             return
 
         dc_fail_until.pop(dc_key, None)
+        ip_fail_until.pop(target, None)
+        ws_pool.report_success(dc, is_media)
         stats.connections_ws += 1
 
         splitter = None
@@ -484,15 +464,15 @@ _client_tasks: Set[asyncio.Task] = set()
 
 
 async def _run(stop_event: Optional[asyncio.Event] = None):
-    global _server_instance, _server_stop_event, fronting_until
+    global _server_instance, _server_stop_event
     _server_stop_event = stop_event
 
     ws_pool.reset()
     cf_worker_pool.reset()
     ws_blacklist.clear()
     dc_fail_until.clear()
+    ip_fail_until.clear()
     _client_tasks.clear()
-    fronting_until = 0.0
 
     if proxy_config.fallback_cfproxy:
         user = proxy_config.cfproxy_user_domains
@@ -691,6 +671,11 @@ def main():
                     metavar='DOMAIN',
                     help='Enable Fake TLS (ee-secret) masking with the given '
                          'SNI domain, e.g. example.com')
+    ap.add_argument('--force-test-dc', action='store_true',
+                    help='Force ALL traffic to Telegram TEST datacenters. '
+                         'Not needed for Telegram Desktop (test DCs 10001+ '
+                         'are detected automatically); use for clients that '
+                         'signal test DCs as plain 1-3')
     ap.add_argument('--proxy-protocol', action='store_true',
                     help='Accept PROXY protocol v1 header '
                          '(for use behind nginx/haproxy with proxy_protocol on)')
@@ -730,6 +715,7 @@ def main():
     proxy_config.cfproxy_worker_domains = coerce_domain_list(args.cfproxy_worker_domain)
     proxy_config.fake_tls_domain = args.fake_tls_domain.strip()
     proxy_config.proxy_protocol = args.proxy_protocol
+    proxy_config.force_test_dc = args.force_test_dc
 
     log_level = logging.DEBUG if args.verbose else logging.INFO
     log_fmt = logging.Formatter('%(asctime)s  %(levelname)-5s  %(message)s',

proxy/utils.py

@@ -43,6 +43,15 @@ DC_DEFAULT_IPS: Dict[int, str] = {
     203: '91.105.192.100'
 }
 
+DC_TEST_IPS: Dict[int, str] = {
+    1: '149.154.175.10',
+    2: '149.154.167.40',
+    3: '149.154.175.117',
+}
+
+WS_PATH = '/apiws'
+WS_PATH_TEST = WS_PATH + '_test'
+
 
 def ws_domains(dc: int, is_media) -> List[str]:
     if dc == 203:

ui/i18n/__init__.py

@@ -25,7 +25,10 @@ class LocaleEnum(str, Enum):
             return _DEFAULT_LOCALE
 
 
-_LOCALES_DIR = Path(__file__).resolve().parent
+try:
+    _LOCALES_DIR = Path(__file__).resolve(strict=False).parent
+except OSError:
+    _LOCALES_DIR = Path(os.path.realpath(__file__)).parent
 _DEFAULT_LOCALE = LocaleEnum.english
 
 _translations: Dict[str, str] = {}

utils/default_config.py

@@ -22,6 +22,7 @@ _TRAY_DEFAULTS_COMMON: Dict[str, Any] = {
     "cfproxy": True,
     "cfproxy_user_domain": [],
     "cfproxy_worker_domain": [],
+    "force_test_dc": False,
     "ws_keepalive_interval": 30,
 }
 

utils/tray_common.py

@@ -41,7 +41,10 @@ def _exe_dir() -> Optional[Path]:
         return None
     if not base:
         return None
-    p = Path(base).resolve()
+    try:
+        p = Path(base).resolve(strict=False)
+    except OSError:
+        p = Path(os.path.realpath(base))
     return p.parent if p.is_file() else p
 
 
@@ -334,6 +337,7 @@ def apply_proxy_config(cfg: dict) -> bool:
     pc.fallback_cfproxy = cfg.get("cfproxy", DEFAULT_CONFIG["cfproxy"])
     pc.cfproxy_user_domains = coerce_domain_list(cfg.get("cfproxy_user_domain", DEFAULT_CONFIG["cfproxy_user_domain"]))
     pc.cfproxy_worker_domains = coerce_domain_list(cfg.get("cfproxy_worker_domain", DEFAULT_CONFIG["cfproxy_worker_domain"]))
+    pc.force_test_dc = cfg.get("force_test_dc", DEFAULT_CONFIG["force_test_dc"])
     pc.ws_keepalive_interval = max(0, cfg.get("ws_keepalive_interval", DEFAULT_CONFIG["ws_keepalive_interval"]))
     return True
 

windows.py

@@ -522,8 +522,10 @@ def _edit_config_dialog() -> None:
                 messagebox.showerror(t("app.error_title"), merged, parent=root)
                 return
 
+            merged["force_test_dc"] = _config.get("force_test_dc", DEFAULT_CONFIG["force_test_dc"])
+
             _ui_only_keys = {"appearance", "autostart", "check_updates", "language"}
-            config_changed = any(merged.get(k) != _config.get(k) for k in merged)
+            config_changed = any(merged.get(k) != cfg.get(k) for k in merged)
             proxy_changed = any(merged.get(k) != _config.get(k) for k in merged if k not in _ui_only_keys)
 
             if not config_changed: