From 88b7b55c58b99d0c6ae4417486b73c79c3fba1cb Mon Sep 17 00:00:00 2001 From: Flowseal Date: Sun, 19 Jul 2026 15:28:57 +0300 Subject: [PATCH] fronting refactoring: implemented into ws_pool --- proxy/pool.py | 16 +++++++++----- proxy/tg_ws_proxy.py | 50 +++----------------------------------------- 2 files changed, 14 insertions(+), 52 deletions(-) diff --git a/proxy/pool.py b/proxy/pool.py index 7a6ef56..d3831c3 100644 --- a/proxy/pool.py +++ b/proxy/pool.py @@ -19,7 +19,6 @@ class _WsPool: def __init__(self): self._idle: Dict[Tuple[int, bool], deque] = {} self._refilling: Set[Tuple[int, bool]] = set() - self.fronting_until: float = 0.0 async def get(self, dc: int, is_media: bool, target_ip: str, domains: List[str] @@ -62,7 +61,7 @@ class _WsPool: if needed <= 0: return 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: @@ -77,11 +76,19 @@ class _WsPool: self._refilling.discard(key) @staticmethod - async def _connect_one(target_ip, domains, fronting_active) -> Optional[RawWebSocket]: + async def _connect_one(target_ip, domains) -> Optional[RawWebSocket]: for domain in domains: try: return await RawWebSocket.connect( - target_ip, domain, timeout=8, sni="sprinthost.ru" if fronting_active else None) + target_ip, domain, timeout=8) + except asyncio.TimeoutError: + try: + ws = await RawWebSocket.connect( + target_ip, domain, timeout=7, sni="sprinthost.ru") + stats.connections_fronting += 1 + return ws + except Exception: + return None except WsHandshakeError as exc: if exc.is_redirect: continue @@ -109,7 +116,6 @@ class _WsPool: def reset(self): self._idle.clear() self._refilling.clear() - self.fronting_until = 0.0 class _CfWorkerPool: diff --git a/proxy/tg_ws_proxy.py b/proxy/tg_ws_proxy.py index 85846bc..51aadd0 100644 --- a/proxy/tg_ws_proxy.py +++ b/proxy/tg_ws_proxy.py @@ -36,13 +36,11 @@ log = logging.getLogger('tg-mtproto-proxy') 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] = {} ip_fail_until: Dict[str, float] = {} -fronting_until: float = 0.0 def _try_handshake(handshake: bytes, secret: bytes) -> Optional[Tuple[int, bool, bytes, bytes]]: @@ -250,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') @@ -336,7 +332,6 @@ async def _handle_client(reader, writer, secret: bytes): return ws_timeout = WS_FAIL_TIMEOUT if now < dc_fail_until.get(dc_key, 0) else 5.0 - fronting_active = now < fronting_until domains = ws_domains(dc, is_media) ws = None @@ -348,25 +343,6 @@ async def _handle_client(reader, writer, secret: bytes): 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=5.0, path=ws_path, - 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}{ws_path}' @@ -403,31 +379,12 @@ async def _handle_client(reader, writer, secret: bytes): 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=5.0, path=ws_path, - 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) @@ -502,7 +459,7 @@ _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() @@ -511,7 +468,6 @@ async def _run(stop_event: Optional[asyncio.Event] = None): 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