From c317406b52c9f54673c3ec135839809db7b46820 Mon Sep 17 00:00:00 2001 From: holdingq0 Date: Sat, 10 Oct 2026 22:36:55 +0300 Subject: [PATCH] =?UTF-8?q?=D0=9F=D1=83=D0=BB=20CF=20worker:=20=D0=B7?= =?UTF-8?q?=D0=B0=D0=BA=D1=80=D1=8B=D1=82=D0=B8=D0=B5=20=D0=BF=D1=80=D0=B8?= =?UTF-8?q?=20=D0=BE=D1=81=D1=82=D0=B0=D0=BD=D0=BE=D0=B2=D0=BA=D0=B5,=20OP?= =?UTF-8?q?TI,=20=D1=81=D0=B1=D0=BE=D1=80=D0=BA=D0=B0,=20=D0=B4=D0=BE?= =?UTF-8?q?=D0=BA=D0=B8=20(#1411)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/EN/BuildFromSource.md | 3 ++- docs/EN/TrayConfig.md | 13 +++++++++---- docs/RU/BuildFromSource.md | 3 ++- docs/RU/TrayConfig.md | 13 +++++++++---- packaging/linux.spec | 2 +- packaging/macos.spec | 2 +- packaging/windows.spec | 2 +- proxy/pool.py | 28 +++++++++++++++++++++++----- proxy/tg_ws_proxy.py | 1 + proxy/utils.py | 3 ++- tests/test_pool.py | 28 +++++++++++++++++++++++++++- 11 files changed, 78 insertions(+), 20 deletions(-) diff --git a/docs/EN/BuildFromSource.md b/docs/EN/BuildFromSource.md index 0b8461d..d382df6 100644 --- a/docs/EN/BuildFromSource.md +++ b/docs/EN/BuildFromSource.md @@ -55,11 +55,12 @@ tg-ws-proxy [--port PORT] [--host HOST] [--dc-ip DC:IP ...] [-v] | `--no-secure` | `false` | Use 80 port for CF-proxy and CF-worker connections | | `--fake-tls-domain` | | Enable Fake TLS masquerading (ee-secret) with specified SNI domain | | `--proxy-protocol` | disabled | Accept HAProxy PROXY protocol v1 (for use behind nginx/haproxy with `proxy_protocol on`) | +| `--force-test-dc` | `false` | Route all traffic to Telegram test DCs [Learn more](./TestDc.md) | | `--buf-kb` | `256` | Buffer size in KB | | `--pool-size` | `4` | Number of ready WS connections per DC. `0` disables the direct DC→IP WS route | | `--log-file` | disabled | Path to file for saving logs | | `--log-max-mb` | `5` | Maximum log file size in MB (afterwards overwrites) | -| `--log-backups` | `0` | Number of log backups after overwrite | +| `--log-backups` | `1` | Number of rotated log files to keep (minimum 1) | | `-v`, `--verbose` | disabled | Verbose logging (DEBUG) | **Examples:** diff --git a/docs/EN/TrayConfig.md b/docs/EN/TrayConfig.md index 68afe8d..7459776 100644 --- a/docs/EN/TrayConfig.md +++ b/docs/EN/TrayConfig.md @@ -18,13 +18,18 @@ The tray application stores data in: "verbose": false, "buf_kb": 256, "pool_size": 4, - "log_max_mb": 5.0, + "log_max_mb": 5, "check_updates": true, "cfproxy": true, - "cfproxy_user_domain": "", - "cfproxy_worker_domain": "", + "h2": true, + "cfproxy_user_domain_enabled": false, + "cfproxy_user_domain": [], + "cfproxy_worker_enabled": false, + "cfproxy_worker_domain": [], "force_test_dc": false, - "appearance": "auto" + "no_secure": false, + "appearance": "auto", + "language": "en" } ``` diff --git a/docs/RU/BuildFromSource.md b/docs/RU/BuildFromSource.md index 6cba6a4..1d3d27d 100644 --- a/docs/RU/BuildFromSource.md +++ b/docs/RU/BuildFromSource.md @@ -55,11 +55,12 @@ tg-ws-proxy [--port PORT] [--host HOST] [--dc-ip DC:IP ...] [-v] | `--no-secure` | `false` | Принудительно использует 80 порт для CF-proxy и CF-worker | | `--fake-tls-domain` | | Включить маскировку Fake TLS (ee-secret) с указанным SNI-доменом | | `--proxy-protocol` | выкл. | Принимать HAProxy PROXY protocol v1 (для работы за nginx/haproxy с `proxy_protocol on`) | +| `--force-test-dc` | `false` | Направлять весь трафик на тестовые DC Telegram [Подробнее](./TestDc.md) | | `--buf-kb` | `256` | Размер буфера в КБ | | `--pool-size` | `4` | Количество заготовленных WS-соединений на каждый DC. `0` отключает прямой WS-маршрут DC→IP | | `--log-file` | выкл. | Путь к файлу, в который будут сохраняться логи | | `--log-max-mb` | `5` | Максимальный размер файла логов в МБ (после этого начинается перезапись) | -| `--log-backups` | `0` | Количество сохранений логов после перезаписи | +| `--log-backups` | `1` | Сколько старых файлов логов хранить при ротации (минимум 1) | | `-v`, `--verbose` | выкл. | Подробное логирование (DEBUG) | **Примеры:** diff --git a/docs/RU/TrayConfig.md b/docs/RU/TrayConfig.md index 89ca46d..6f177fd 100644 --- a/docs/RU/TrayConfig.md +++ b/docs/RU/TrayConfig.md @@ -18,13 +18,18 @@ Tray-приложение хранит данные в: "verbose": false, "buf_kb": 256, "pool_size": 4, - "log_max_mb": 5.0, + "log_max_mb": 5, "check_updates": true, "cfproxy": true, - "cfproxy_user_domain": "", - "cfproxy_worker_domain": "", + "h2": true, + "cfproxy_user_domain_enabled": false, + "cfproxy_user_domain": [], + "cfproxy_worker_enabled": false, + "cfproxy_worker_domain": [], "force_test_dc": false, - "appearance": "auto" + "no_secure": false, + "appearance": "auto", + "language": "ru" } ``` diff --git a/packaging/linux.spec b/packaging/linux.spec index 3bf72ee..745ef52 100644 --- a/packaging/linux.spec +++ b/packaging/linux.spec @@ -24,7 +24,7 @@ a = Analysis( [os.path.join(os.path.dirname(SPEC), os.pardir, 'linux.py')], pathex=[], binaries=appindicator_binaries, - datas=[(ctk_path, 'customtkinter/'), (_i18n_path, 'ui/i18n')], + datas=[(os.path.join(ctk_path, 'assets'), 'customtkinter/assets'), (_i18n_path, 'ui/i18n')], hiddenimports=[ 'httpx', 'h2', diff --git a/packaging/macos.spec b/packaging/macos.spec index efbea71..8cfe00f 100644 --- a/packaging/macos.spec +++ b/packaging/macos.spec @@ -15,7 +15,7 @@ a = Analysis( [os.path.join(os.path.dirname(SPEC), os.pardir, 'macos.py')], pathex=[], binaries=[], - datas=[(ctk_path, 'customtkinter/'), (_i18n_path, 'ui/i18n')] + certifi_datas, + datas=[(os.path.join(ctk_path, 'assets'), 'customtkinter/assets'), (_i18n_path, 'ui/i18n')] + certifi_datas, hiddenimports=[ 'httpx', 'h2', diff --git a/packaging/windows.spec b/packaging/windows.spec index e9e8800..38041f5 100644 --- a/packaging/windows.spec +++ b/packaging/windows.spec @@ -18,7 +18,7 @@ a = Analysis( [os.path.join(os.path.dirname(SPEC), os.pardir, 'windows.py')], pathex=[], binaries=[], - datas=[(ctk_path, 'customtkinter/'), (_i18n_path, 'ui/i18n')] + certifi_datas, + datas=[(os.path.join(ctk_path, 'assets'), 'customtkinter/assets'), (_i18n_path, 'ui/i18n')] + certifi_datas, hiddenimports=[ 'httpx', 'h2', diff --git a/proxy/pool.py b/proxy/pool.py index 824671e..7f6d1ea 100644 --- a/proxy/pool.py +++ b/proxy/pool.py @@ -5,7 +5,7 @@ import time from collections import deque from urllib.parse import urlencode -from typing import Dict, List, Optional, Tuple, Set +from typing import Dict, List, Optional, Tuple from .raw_websocket import RawWebSocket, WsHandshakeError from .stats import stats @@ -237,7 +237,7 @@ class _CfWorkerPool: def __init__(self): self._idle: Dict[int, deque] = {} - self._refilling: Set[int] = set() + self._refilling: Dict[int, asyncio.Task] = {} self._exhausted_until: Dict[str, float] = {} async def get(self, dc: int, fallback_dst: str, @@ -269,8 +269,7 @@ class _CfWorkerPool: def _schedule_refill(self, dc, fallback_dst, worker_domains): if dc in self._refilling: return - self._refilling.add(dc) - asyncio.create_task(self._refill( + self._refilling[dc] = asyncio.create_task(self._refill( dc, fallback_dst, list(worker_domains))) async def _refill(self, dc, fallback_dst, worker_domains): @@ -287,11 +286,15 @@ class _CfWorkerPool: if connected is None: break ws, worker_domain = connected + if self._refilling.get(dc) is not asyncio.current_task(): + await self._quiet_close(ws) + return bucket.append((ws, time.monotonic(), worker_domain)) log.debug("CF worker pool refilled DC%d: %d ready", dc, len(bucket)) finally: - self._refilling.discard(dc) + if self._refilling.get(dc) is asyncio.current_task(): + self._refilling.pop(dc, None) async def _connect_one(self, worker_domains, fallback_dst, dc): query = urlencode({ @@ -359,10 +362,25 @@ class _CfWorkerPool: log.info("CF worker pool warmup started for %d DC(s)", len(cf_fallbacks)) def reset(self): + loop = asyncio.get_running_loop() + for task in self._refilling.values(): + if not task.done() and task.get_loop() is loop: + task.cancel() + for bucket in self._idle.values(): + for ws, _, _ in bucket: + try: + ws.writer.close() + except Exception as exc: + log.debug("CF worker pool close failed: %r", exc) self._idle.clear() self._refilling.clear() self._exhausted_until.clear() + async def close(self): + tasks = list(self._refilling.values()) + self.reset() + await asyncio.gather(*tasks, return_exceptions=True) + ws_pool = _WsPool() cf_worker_pool = _CfWorkerPool() diff --git a/proxy/tg_ws_proxy.py b/proxy/tg_ws_proxy.py index 34e3f5f..d81baf7 100644 --- a/proxy/tg_ws_proxy.py +++ b/proxy/tg_ws_proxy.py @@ -536,6 +536,7 @@ async def _run(stop_event: Optional[asyncio.Event] = None): await _quiet_cancel(task) await server.wait_closed() await ws_pool.close() + await cf_worker_pool.close() if cf_h2_pool is not None: await cf_h2_pool.close() cf_h2_pool = None diff --git a/proxy/utils.py b/proxy/utils.py index 34c9097..930965e 100644 --- a/proxy/utils.py +++ b/proxy/utils.py @@ -33,7 +33,8 @@ 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'} + b'\xdd\xdd\xdd\xdd', b'\x16\x03\x01\x02', + b'\x4F\x50\x54\x49'} RESERVED_CONTINUE = b'\x00\x00\x00\x00' _GITHUB_IPS: Dict[str, str] = { diff --git a/tests/test_pool.py b/tests/test_pool.py index 6973ee3..38c521e 100644 --- a/tests/test_pool.py +++ b/tests/test_pool.py @@ -7,7 +7,7 @@ from types import SimpleNamespace from unittest import mock from proxy.config import proxy_config -from proxy.pool import _WsPool +from proxy.pool import _CfWorkerPool, _WsPool from proxy.raw_websocket import RawWebSocket, WsHandshakeError from proxy.utils import ws_domains @@ -241,5 +241,31 @@ class WsPoolTest(unittest.IsolatedAsyncioTestCase): idle.writer.close.assert_called_once() +class CfWorkerPoolTest(unittest.IsolatedAsyncioTestCase): + async def test_close_cancels_refill_and_closes_unclaimed_connections(self): + pool = _CfWorkerPool() + ready = _open_ws() + idle = _open_ws() + started = asyncio.Event() + pool._idle[2] = deque([(idle, time.monotonic(), 'a.example')]) + + async def connect(*args): + started.set() + try: + await asyncio.Future() + except asyncio.CancelledError: + return ready, 'b.example' + + with mock.patch.object(proxy_config, 'pool_size', 1), \ + mock.patch.object(pool, '_connect_one', side_effect=connect): + pool._schedule_refill(4, '192.0.2.4', ['b.example']) + await asyncio.wait_for(started.wait(), 1) + await pool.close() + self.assertFalse(pool._idle) + self.assertFalse(pool._refilling) + ready.writer.close.assert_called_once() + idle.writer.close.assert_called_once() + + if __name__ == '__main__': unittest.main()