Version bump

.github/scripts/answers/default.md

@@ -0,0 +1,23 @@
+### Проверьте две вещи:
+- вы на последней версии: [Releases](https://github.com/Flowseal/tg-ws-proxy/releases)
+- запускали по инструкции для своей ОС: [Быстрый старт](https://github.com/Flowseal/tg-ws-proxy#навигация)
+
+## Решение частых проблем:
+
+**Q**: Не запускается, падает с ошибкой, не работает как раньше после обновления?  
+**A**:
+1. Удалите всё в папке Temp (или хотя бы всё, что начинается с _MEI) 
+2. Запускайте от имени админа
+3. Попробуйте Win7 версию (если вы пользователь Windows)
+4. Попробуйте отключить антивирус (если помогло, то добавьте exe в исключения). Не забудьте включить антивирус обратно.
+
+###
+
+**Q**: Не грузит медиа? (фото/видео/стикеры)  
+**A**: Удалите в настройках прокси в поле **DC → IP** всё, кроме `4:149.154.167.220`. Если это не помогло, полностью очистите это поле.
+
+
+#### Если проблема решена, то закройте Issue
+
+### Если проблема осталась, пожалуйста, приложите по возможности логи.
+Сделать это можно через иконку в трее -> Пкм -> Открыть логи. Сохраните логи в файл и приложите его сюда.

.github/scripts/answers/no_cfproxy.md

@@ -0,0 +1,10 @@
+## У вас не включен CF-прокси, включите его в настройках.
+
+Если в логах будет **TimeoutError**, то:
+
+Timeout error при CF-прокси - проблема с вашей стороны. РКН/сторонние впны/прокси и т.п.
+
+1. Попробуйте поставить галочку "Выключить TLS для CF-прокси и CF-worker" в настройках. Если не поможет, уберите.  
+2. Если используете запрет, попробуйте его выключить и проверьте работу. Если работает, то добавьте в файл `list-exclude-user`: **`co.uk`**
+3. Если не помогло, то уберите запись из прошлого пункта и добавьте в файл `list-general-user`: **`co.uk`**
+4. Попробуйте настроить https://github.com/Flowseal/tg-ws-proxy/blob/main/docs/RU/CfWorker.md

.github/scripts/answers/reset_error.md

@@ -0,0 +1,4 @@
+Проверьте включенли WARP/VPN, если выключены то: 
+
+1. Если используете запрет, добавьте в файл `list-general-user`: **`co.uk`**
+2. Попробуйте настроить https://github.com/Flowseal/tg-ws-proxy/blob/main/docs/RU/CfWorker.md

.github/scripts/answers/timeout.md

@@ -0,0 +1,5 @@
+Timeout error при CF-прокси - проблема с вашей стороны. РКН/сторонние впны/прокси и т.п. 
+
+1. Попробуйте поставить галочку "Выключить TLS для CF-прокси и CF-worker" в настройках. Если не поможет, уберите.  
+2. Если используете запрет, попробуйте его выключить и проверьте работу. Если работает, то добавьте в файл `list-exclude-user`: **`co.uk`**
+3. Если не помогло, то уберите запись из прошлого пункта и добавьте в файл `list-general-user`: **`co.uk`**

.github/scripts/answers/timeout_no_worker.md

@@ -0,0 +1,6 @@
+Timeout error при CF-прокси - проблема с вашей стороны. РКН/сторонние впны/прокси и т.п. 
+
+1. Попробуйте поставить галочку "Выключить TLS для CF-прокси и CF-worker" в настройках. Если не поможет, уберите.  
+2. Если используете запрет, попробуйте его выключить и проверьте работу. Если работает, то добавьте в файл `list-exclude-user`: **`co.uk`**
+3. Если не помогло, то уберите запись из прошлого пункта и добавьте в файл `list-general-user`: **`co.uk`**
+4. Попробуйте настроить https://github.com/Flowseal/tg-ws-proxy/blob/main/docs/RU/CfWorker.md

.github/scripts/answers/wrong_secret.md

@@ -0,0 +1,3 @@
+Указан **неверный секрет** в настройках прокси в телеграме
+
+`Лкм` по иконке в трее -> **Открыть в Telegram**

.github/scripts/check_log.py

@@ -0,0 +1,65 @@
+import os
+from time import sleep
+import urllib.request as request
+
+url = os.environ["LOG_URL"]
+path_answer = ".github/scripts/answers"
+
+for _ in range(0,3):
+    try:
+        with request.urlopen(url,timeout=30.0) as r:
+            log_bytes = r.read()
+        break 
+    except Exception:
+        sleep(5.0)
+else:
+    print(str(path_answer + "/default.md"))
+    exit()
+
+log_old = log_bytes.decode("utf-8",errors="replace").splitlines()
+
+worker_end = False
+cfproxy_end = False
+
+log = []
+for line in reversed(log_old):
+    line = line[20:]
+    log.append(line)
+    if "CF proxy:      enabled" in line:
+        cfproxy_end = True
+    elif "CF worker:     enabled" in line:
+        worker_end = True
+    elif "INFO   tg-ws-tray  Starting proxy" in line:
+        break
+
+secret_false = 0
+timeouterror = 0
+reset_error = 0 
+
+for line in log :
+    if "(wrong secret or proto)" in line:
+        secret_false += 1 
+        if secret_false >= 10:
+            print(str(path_answer + "/wrong_secret.md"))
+            break
+    elif "TimeoutError" in line: 
+        timeouterror += 1 
+        if timeouterror >= 20:
+            if cfproxy_end is False:
+                print(str(path_answer + "/no_cfproxy.md"))
+                break
+            if worker_end is False:
+                print(str(path_answer + "/timeout_no_worker.md"))
+            else:
+                print(str(path_answer + "/timeout.md"))                  
+            break
+    elif "ConnectionResetError" in line:
+        reset_error += 1 
+        if reset_error >= 20:
+            print(str(path_answer + "/reset_error.md"))
+            break
+else: 
+    if cfproxy_end is True:
+        print(str(path_answer + "/default.md"))
+    else:
+        print(str(path_answer + "/no_cfproxy.md"))

.github/workflows/build.yml

@@ -664,7 +664,7 @@ jobs:
 
             > [!TIP]
             > Не можете скачать?
-            > Добавьте `185.199.109.133 release-assets.githubusercontent.com` в hosts или воспользуйтесь зеркалом: https://sourceforge.net/projects/tg-ws-proxy.mirror/files/
+            > Воспользуйтесь зеркалом: https://sourceforge.net/projects/tg-ws-proxy.mirror/files/
           files: |
             dist/TgWsProxy_windows.exe
             dist/TgWsProxy_windows_arm64.exe

.github/workflows/issue-triage.yml

@@ -6,37 +6,54 @@ on:
 
 permissions:
   issues: write
+  contents: read
 
 jobs:
   comment:
     if: contains(github.event.issue.labels.*.name, 'bug')
     runs-on: ubuntu-latest
     steps:
+      - name: Check for availability log file
+        id: log
+        env: 
+          ISSUE_BODY: ${{ github.event.issue.body }}
+        run: |
+          if [[ "$ISSUE_BODY" =~ (https://github\.com/user-attachments/files/[^[:space:]]+\.log) ]]; then
+            echo "found=true" >> "$GITHUB_OUTPUT"
+            echo "url=${BASH_REMATCH[1]}" >> "$GITHUB_OUTPUT"
+            echo "Log file found"
+          else
+            echo "found=false" >> "$GITHUB_OUTPUT"
+            echo "Log file not found"
+          fi
+
+      - name: Install python
+        if: steps.log.outputs.found == 'true'
+        uses: actions/setup-python@v6
+        with:
+          python-version: "3.14"
+
+      - name: Clone folder 
+        uses: actions/checkout@v4
+        with:
+          sparse-checkout: |
+            .github/scripts/*
+          sparse-checkout-cone-mode: false
+
+      - name: Start check log
+        if: steps.log.outputs.found == 'true'
+        id: comment
+        env:
+          LOG_URL: ${{ steps.log.outputs.url }}
+        run: |
+          {
+            echo "com<<EOF"
+            python .github/scripts/check_log.py
+            echo "EOF"
+          } >> "$GITHUB_OUTPUT"
+
       - name: Comment on new issue
         uses: peter-evans/create-or-update-comment@v5
         with:
           issue-number: ${{ github.event.issue.number }}
-          body: |
-            ### Проверьте две вещи:
-            - вы на последней версии: [Releases](https://github.com/Flowseal/tg-ws-proxy/releases)
-            - запускали по инструкции для своей ОС: [Быстрый старт](https://github.com/Flowseal/tg-ws-proxy#навигация)
-
-            ## Решение частых проблем:
-
-            **Q**: Не запускается, падает с ошибкой, не работает как раньше после обновления?  
-            **A**:
-            1. Удалите всё в папке Temp (или хотя бы всё, что начинается с _MEI) 
-            2. Запускайте от имени админа
-            3. Попробуйте Win7 версию (если вы пользователь Windows)
-            4. Попробуйте отключить антивирус (если помогло, то добавьте exe в исключения). Не забудьте включить антивирус обратно.
-
-            ###
-
-            **Q**: Не грузит медиа? (фото/видео/стикеры)  
-            **A**: Удалите в настройках прокси в поле **DC → IP** всё, кроме `4:149.154.167.220`. Если это не помогло, полностью очистите это поле.
-
-
-            #### Если проблема решена, то закройте Issue
-
-            ### Если проблема осталась, пожалуйста, приложите по возможности логи.
-            Сделать это можно через иконку в трее -> Пкм -> Открыть логи. Сохраните логи в файл и приложите его сюда.
+          body-path: ${{ steps.log.outputs.found == 'true' && steps.comment.outputs.com || '.github/scripts/answers/default.md' }}

Dockerfile

@@ -26,7 +26,8 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
     TG_WS_PROXY_PORT=1443 \
     TG_WS_PROXY_SECRET=""  \
     TG_WS_PROXY_DC_IPS="2:149.154.167.220 4:149.154.167.220" \
-    TG_WS_PROXY_CF_WORKER=""
+    TG_WS_PROXY_CF_WORKER="" \
+    TG_WS_PROXY_NO_SECURE=False
 
 RUN apt-get update \
     && apt-get install -y --no-install-recommends tini ca-certificates \
@@ -44,5 +45,17 @@ USER app
 
 EXPOSE 1443/tcp
 
-ENTRYPOINT ["/usr/bin/tini", "--", "/bin/sh", "-lc", "set -eu; args=\"--host ${TG_WS_PROXY_HOST} --port ${TG_WS_PROXY_PORT}\"; for dc in ${TG_WS_PROXY_DC_IPS}; do args=\"$args --dc-ip $dc\"; done; if [ -n \"${TG_WS_PROXY_SECRET}\" ]; then args=\"$args --secret ${TG_WS_PROXY_SECRET}\"; fi; if [ -n \"${TG_WS_PROXY_CF_WORKER}\" ]; then args=\"$args --cfproxy-worker-domain ${TG_WS_PROXY_CF_WORKER}\"; fi; exec /opt/venv/bin/python -u proxy/tg_ws_proxy.py $args \"$@\"", "--"]
+ENTRYPOINT ["/usr/bin/tini", "--", "/bin/sh", "-lc", "set -eu; args=\"\
+    --host ${TG_WS_PROXY_HOST} \
+    --port ${TG_WS_PROXY_PORT}\"; \
+    for dc in ${TG_WS_PROXY_DC_IPS}; \
+        do args=\"$args --dc-ip $dc\"; \
+    done; \
+    if [ -n \"${TG_WS_PROXY_SECRET}\" ]; \
+        then args=\"$args --secret ${TG_WS_PROXY_SECRET}\"; fi; \
+    if [ -n \"${TG_WS_PROXY_CF_WORKER}\" ]; \
+        then args=\"$args --cfproxy-worker-domain ${TG_WS_PROXY_CF_WORKER}\"; fi; \
+    if [ \"${TG_WS_PROXY_NO_SECURE}\" = \"True\" ]; \ 
+        then args=\"$args --no-secure \"; fi; \
+    exec /opt/venv/bin/python -u proxy/tg_ws_proxy.py $args \"$@\"", "--"]
 CMD []

docs/EN/README.docker.md

@@ -36,12 +36,13 @@ tg://proxy?server=172.17.0.2&port=1443&secret=dd68f127db1d...
 All settings are configured using environment variables when running the container:
 
 | Variable                | Description                                        | Default                               |
-| ----------------------- | -------------------------------- | --------------------------------- |
+| ----------------------- | -------------------------------------------------- | ------------------------------------- |
 | `TG_WS_PROXY_HOST`      | Address for incoming connections                   | `0.0.0.0`                             |
 | `TG_WS_PROXY_PORT`      | Port inside the container                          | `1443`                                |
 | `TG_WS_PROXY_SECRET`    | Secret key                                         | `random`                              |
 | `TG_WS_PROXY_DC_IPS`    | DC number:IP pairs separated by space              | `2:149.154.167.220 4:149.154.167.220` |
 | `TG_WS_PROXY_CF_WORKER` | Cloudflare Worker domain                           | `None`                                |
+| `TG_WS_PROXY_NO_SECURE` | Use 80 port for CF-proxy and CF-worker connections | `False`                               |
 
 Example with manually specified secret:
 

docs/RU/README.docker.md

@@ -36,12 +36,13 @@ tg://proxy?server=172.17.0.2&port=1443&secret=dd68f127db1d...
 Все настройки задаются переменными окружения при запуске контейнера:
 
 | Переменная              | Описание                                                      | По умолчанию                          |
-| ----------------------- | --------------------------------- | ------------------------------------- |
+| ----------------------- | ------------------------------------------------------------- | ------------------------------------- |
 | `TG_WS_PROXY_HOST`      | `Адрес для приёма подключений`                                | `0.0.0.0`                             |
 | `TG_WS_PROXY_PORT`      | `Порт внутри контейнера`                                      | `1443`                                |
 | `TG_WS_PROXY_SECRET`    | `Секретный ключ`                                              | `random`                              |
 | `TG_WS_PROXY_DC_IPS`    | `Пары «номер DC:IP» через пробел`                             | `2:149.154.167.220 4:149.154.167.220` |
 | `TG_WS_PROXY_CF_WORKER` | `Домен Cloudflare Worker`                                     | `None`                                |
+| `TG_WS_PROXY_NO_SECURE` | `Использовать 80 порт для подключения к CF-proxy и CF-Worker` | `False`                               |
 
 Пример с ручным указанием секрета:
 

packaging/version_info.txt

@@ -4,8 +4,8 @@
 # http://msdn.microsoft.com/en-us/library/ms646997.aspx
 VSVersionInfo(
   ffi=FixedFileInfo(
-    filevers=(1, 11, 2, 0),
-    prodvers=(1, 11, 2, 0),
+    filevers=(1, 11, 4, 0),
+    prodvers=(1, 11, 4, 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.11.2.0'),
+            StringStruct(u'FileVersion', u'1.11.4.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.11.2.0'),
+            StringStruct(u'ProductVersion', u'1.11.4.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.11.2"
+__version__ = "1.11.4"
 
 __all__ = ["__version__", "get_link_host", "proxy_config", "parse_dc_ip_list", "build_github_opener", "coerce_domain_list"]

proxy/fake_tls.py

@@ -5,12 +5,19 @@ import hmac
 import hashlib
 import os
 import random
+import ssl
 import struct
 import time
 import logging
 
-from typing import Optional, Tuple
+from typing import Dict, FrozenSet, NamedTuple, Optional, Tuple
 from .stats import stats
+from .utils import set_x25519
+
+try:
+    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+except ImportError:
+    X25519PrivateKey = None
 
 
 log = logging.getLogger('tg-mtproto-proxy')
@@ -32,26 +39,46 @@ TIMESTAMP_TOLERANCE = 120
 
 TLS_APPDATA_MAX = 16384
 
+EXT_SUPPORTED_VERSIONS = 0x002b
+EXT_KEY_SHARE = 0x0033
+
+GROUP_X25519 = 0x001d
+GROUP_X25519MLKEM768 = 0x11ec
+
+X25519_KEY_LEN = 32
+X25519MLKEM768_CLIENT_SHARE_LEN = 1216
+MLKEM768_CIPHERTEXT_LEN = 1088
+
+# TLS 1.3 suites in the order a typical server picks them from the client list
+CIPHER_PREFERENCE = (0x1301, 0x1303, 0x1302)
+
+# Bounds for the single ApplicationData record that follows CCS; the low bound
+# matches the smallest record telemt emits, the high one is the TLS limit
+APPDATA_MIN = 64
+# Without a site profile stay near the 64-byte record telemt passed DPI with,
+# far from the 1900-2100 bytes of the classic MTProxy answer
+FALLBACK_APPDATA_RANGE = (64, 160)
+
+SERVER_HELLO_DELAY = (0.008, 0.024)
+
+PROFILE_PROBE_TIMEOUT = 5
+
 
 _CCS_FRAME = b'\x14\x03\x03\x00\x01\x01'
 
-_SERVER_HELLO_TEMPLATE = bytearray(
-    b'\x16\x03\x03\x00\x7a'
-    b'\x02\x00\x00\x76'
-    b'\x03\x03'
-    + b'\x00' * 32
-    + b'\x20'
-    + b'\x00' * 32
-    + b'\x13\x01\x00'
-    + b'\x00\x2e'
-    + b'\x00\x33\x00\x24\x00\x1d\x00\x20'
-    + b'\x00' * 32
-    + b'\x00\x2b\x00\x02\x03\x04'
-)
-
 _SH_RANDOM_OFF = 11
-_SH_SESSID_OFF = 44
-_SH_PUBKEY_OFF = 89
+
+# Size of the first encrypted record the masking site sends after ServerHello;
+# set once at startup by probe_appdata_len()
+_profiled_appdata_len: Optional[int] = None
+
+
+class ClientHelloInfo(NamedTuple):
+    cipher_suites: FrozenSet[int]
+    key_shares: Dict[int, int]  # named group -> key_exchange length
+
+
+_EMPTY_HELLO_INFO = ClientHelloInfo(frozenset(), {})
 
 
 def verify_client_hello(data: bytes, secret: bytes) -> Optional[Tuple[bytes, bytes, int]]:
@@ -88,17 +115,114 @@ def verify_client_hello(data: bytes, secret: bytes) -> Optional[Tuple[bytes, byt
     return client_random, session_id, timestamp
 
 
-def build_server_hello(secret: bytes, client_random: bytes, session_id: bytes) -> bytes:
-    sh = bytearray(_SERVER_HELLO_TEMPLATE)
-    sh[_SH_SESSID_OFF:_SH_SESSID_OFF + 32] = session_id
-    sh[_SH_PUBKEY_OFF:_SH_PUBKEY_OFF + 32] = os.urandom(32)
+def parse_client_hello(data: bytes) -> ClientHelloInfo:
+    """Extract offered cipher suites and key_share groups from a ClientHello.
 
-    ccs = _CCS_FRAME
-    encrypted_size = random.randint(1900, 2100)
-    encrypted_data = os.urandom(encrypted_size)
-    app_record = b'\x17\x03\x03' + struct.pack('>H', encrypted_size) + encrypted_data
+    Returns empty info on malformed input so the caller falls back to the
+    plain X25519 / TLS_AES_128_GCM_SHA256 answer.
+    """
+    try:
+        pos = 5 + 4 + 2 + CLIENT_RANDOM_LEN
+        pos += 1 + data[pos]
 
-    response = bytes(sh) + ccs + app_record
+        suites_len = struct.unpack_from('>H', data, pos)[0]
+        pos += 2
+        suites_end = pos + suites_len
+        if suites_end > len(data):
+            return _EMPTY_HELLO_INFO
+        suites = frozenset(
+            struct.unpack_from('>H', data, off)[0]
+            for off in range(pos, suites_end - 1, 2))
+        pos = suites_end
+
+        pos += 1 + data[pos]
+
+        ext_total = struct.unpack_from('>H', data, pos)[0]
+        pos += 2
+        ext_end = min(pos + ext_total, len(data))
+        key_shares: Dict[int, int] = {}
+        while pos + 4 <= ext_end:
+            ext_type, ext_len = struct.unpack_from('>HH', data, pos)
+            pos += 4
+            if ext_type == EXT_KEY_SHARE and pos + 2 <= ext_end:
+                shares_end = min(pos + 2 + struct.unpack_from('>H', data, pos)[0],
+                                 ext_end)
+                off = pos + 2
+                while off + 4 <= shares_end:
+                    group, key_len = struct.unpack_from('>HH', data, off)
+                    off += 4 + key_len
+                    if off <= shares_end:
+                        key_shares.setdefault(group, key_len)
+            pos += ext_len
+    except (IndexError, struct.error):
+        return _EMPTY_HELLO_INFO
+
+    return ClientHelloInfo(suites, key_shares)
+
+
+def _select_cipher_suite(info: ClientHelloInfo) -> int:
+    for suite in CIPHER_PREFERENCE:
+        if suite in info.cipher_suites:
+            return suite
+    return CIPHER_PREFERENCE[0]
+
+
+def _x25519_public_key() -> bytes:
+    if X25519PrivateKey is not None:
+        return X25519PrivateKey.generate().public_key().public_bytes_raw()
+    # Real public keys are field elements below 2^255; keep the top bit clear
+    key = bytearray(os.urandom(X25519_KEY_LEN))
+    key[-1] &= 0x7f
+    return bytes(key)
+
+
+def _server_key_share(info: ClientHelloInfo) -> Tuple[int, bytes]:
+    # Mirror the hybrid post-quantum group when the client offers it, as
+    # current browsers and servers do; the client never derives keys from it
+    if info.key_shares.get(GROUP_X25519MLKEM768) == X25519MLKEM768_CLIENT_SHARE_LEN:
+        return GROUP_X25519MLKEM768, (os.urandom(MLKEM768_CIPHERTEXT_LEN)
+                                      + _x25519_public_key())
+    return GROUP_X25519, _x25519_public_key()
+
+
+def _appdata_len() -> int:
+    if _profiled_appdata_len is not None:
+        return _profiled_appdata_len
+    return random.randint(*FALLBACK_APPDATA_RANGE)
+
+
+def build_server_hello(secret: bytes, client_random: bytes, session_id: bytes,
+                       info: ClientHelloInfo = _EMPTY_HELLO_INFO) -> bytes:
+    """ServerHello + CCS + one ApplicationData record.
+
+    Telegram clients parse exactly these three records and check the HMAC
+    over all of them, so the shape must stay fixed while sizes may vary.
+    """
+    group, key_exchange = _server_key_share(info)
+    extensions = (
+        struct.pack('>HHH', EXT_SUPPORTED_VERSIONS, 2, 0x0304)
+        + struct.pack('>HHHH', EXT_KEY_SHARE, 4 + len(key_exchange),
+                      group, len(key_exchange))
+        + key_exchange
+    )
+
+    body = (
+        TLS_VERSION_12
+        + b'\x00' * 32
+        + bytes([len(session_id)]) + session_id
+        + struct.pack('>H', _select_cipher_suite(info))
+        + b'\x00'
+        + struct.pack('>H', len(extensions)) + extensions
+    )
+    message = b'\x02' + struct.pack('>I', len(body))[1:] + body
+    server_hello = (bytes([TLS_RECORD_HANDSHAKE]) + TLS_VERSION_12
+                    + struct.pack('>H', len(message)) + message)
+
+    appdata_len = _appdata_len()
+    app_record = (bytes([TLS_RECORD_APPDATA]) + TLS_VERSION_12
+                  + struct.pack('>H', appdata_len) + os.urandom(appdata_len))
+
+    response = server_hello + _CCS_FRAME + app_record
 
     hmac_input = client_random + response
     server_random = hmac.new(secret, hmac_input, hashlib.sha256).digest()
@@ -109,6 +233,89 @@ def build_server_hello(secret: bytes, client_random: bytes, session_id: bytes) -
     return bytes(final)
 
 
+def set_profiled_appdata_len(length: Optional[int]) -> None:
+    global _profiled_appdata_len
+    if length is None:
+        _profiled_appdata_len = None
+    else:
+        _profiled_appdata_len = max(APPDATA_MIN, min(length, TLS_APPDATA_MAX))
+
+
+def first_appdata_record_len(data: bytes) -> Optional[int]:
+    """Length of the first ApplicationData record in a server's first flight.
+
+    Returns None while more data is needed, 0 if the flight cannot contain
+    one (alert, TLS 1.2 handshake, garbage).
+    """
+    pos = 0
+    while pos + 5 <= len(data):
+        rtype = data[pos]
+        rec_len = struct.unpack_from('>H', data, pos + 3)[0]
+        if rtype == TLS_RECORD_APPDATA:
+            return rec_len
+        if rtype not in (TLS_RECORD_HANDSHAKE, TLS_RECORD_CCS):
+            return 0
+        pos += 5 + rec_len
+    return None
+
+
+async def probe_appdata_len(domain: str, port: int = 443,
+                            timeout: float = PROFILE_PROBE_TIMEOUT) -> Optional[int]:
+    """Measure the first encrypted record the masking site sends after ServerHello.
+
+    Only the raw first flight is read; the TLS handshake is never completed.
+    """
+    if not getattr(ssl, 'HAS_TLSv1_3', False):
+        log.warning("Fake TLS profile: local ssl lacks TLS 1.3, "
+                    "using random record sizes")
+        return None
+
+    writer = None
+    try:
+        ctx = ssl.create_default_context()
+        set_x25519(ctx)
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+        ctx.set_alpn_protocols(['h2', 'http/1.1'])
+        outgoing = ssl.MemoryBIO()
+        tls = ctx.wrap_bio(ssl.MemoryBIO(), outgoing, server_hostname=domain)
+        try:
+            tls.do_handshake()
+        except ssl.SSLWantReadError:
+            pass
+        client_hello = outgoing.read()
+
+        # Share one deadline across connecting, sending, and reading the flight.
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + timeout
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(domain, port), timeout=timeout)
+        writer.write(client_hello)
+        await asyncio.wait_for(writer.drain(), timeout=deadline - loop.time())
+
+        flight = bytearray()
+        while True:
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                raise asyncio.TimeoutError
+            chunk = await asyncio.wait_for(
+                reader.read(16384), timeout=remaining)
+            if not chunk:
+                return None
+            flight.extend(chunk)
+            length = first_appdata_record_len(bytes(flight))
+            if length is not None:
+                return length or None
+    except Exception as exc:
+        log.warning("Fake TLS profile: cannot probe %s:%d: %r",
+                    domain, port, exc)
+        return None
+    finally:
+        if writer is not None:
+            writer.close()
+
+
 def wrap_tls_record(data: bytes) -> bytes:
     parts = []
     offset = 0

proxy/pool.py

@@ -233,16 +233,24 @@ class _WsPool:
 
 class _CfWorkerPool:
     WS_POOL_MAX_AGE = 100.0
+    WS_POOL_CHECK_INTERVAL = 5.0
     PER_DC_LIMIT = 1
+    REFILL_BACKOFF_INITIAL = 1.0
+    REFILL_BACKOFF_MAX = 3600.0
 
     def __init__(self):
         self._idle: Dict[int, deque] = {}
         self._refilling: Dict[int, asyncio.Task] = {}
+        self._rotating: Dict[int, asyncio.Task] = {}
+        self._refill_failures: Dict[int, int] = {}
+        self._refill_after: Dict[int, float] = {}
         self._exhausted_until: Dict[str, float] = {}
 
     async def get(self, dc: int, fallback_dst: str,
                   worker_domains: List[str]
                   ) -> Optional[Tuple[RawWebSocket, str]]:
+        if proxy_config.pool_size <= 0 or not worker_domains:
+            return None
         now = time.monotonic()
 
         bucket = self._idle.get(dc)
@@ -252,8 +260,7 @@ class _CfWorkerPool:
         while bucket:
             ws, created, worker_domain = bucket.popleft()
             age = now - created
-            if (age > self.WS_POOL_MAX_AGE or ws._closed
-                    or ws.writer.transport.is_closing()):
+            if self._is_stale(ws, created, now):
                 asyncio.create_task(self._quiet_close(ws))
                 continue
             stats.cf_pool_hits += 1
@@ -264,10 +271,20 @@ class _CfWorkerPool:
             return ws, worker_domain
 
         stats.cf_pool_misses += 1
+        self._schedule_refill(dc, fallback_dst, worker_domains)
         return None
 
+    def _is_stale(self, ws, created, now):
+        return (now - created >= self.WS_POOL_MAX_AGE or ws._closed
+                or ws.writer.transport.is_closing() or ws.reader.at_eof()
+                or ws.reader.exception() is not None)
+
     def _schedule_refill(self, dc, fallback_dst, worker_domains):
-        if dc in self._refilling:
+        if proxy_config.pool_size <= 0 or not worker_domains:
+            return
+        self._schedule_rotation(dc, fallback_dst, worker_domains)
+        if (dc in self._refilling
+                or time.monotonic() < self._refill_after.get(dc, 0)):
             return
         self._refilling[dc] = asyncio.create_task(self._refill(
             dc, fallback_dst, list(worker_domains)))
@@ -279,23 +296,80 @@ class _CfWorkerPool:
             needed = target_size - len(bucket)
             if needed <= 0:
                 return
-
+            connected = 0
             for _ in range(needed):
-                connected = await self._connect_one(
+                result = await self._connect_one(
                     worker_domains, fallback_dst, dc)
-                if connected is None:
-                    break
-                ws, worker_domain = connected
                 if self._refilling.get(dc) is not asyncio.current_task():
-                    await self._quiet_close(ws)
+                    if result is not None:
+                        await self._quiet_close(result[0])
                     return
+                if result is None:
+                    break
+                ws, worker_domain = result
                 bucket.append((ws, time.monotonic(), worker_domain))
+                connected += 1
+            if connected:
+                self._refill_failures.pop(dc, None)
+                self._refill_after.pop(dc, None)
+            else:
+                failures = self._refill_failures.get(dc, 0) + 1
+                self._refill_failures[dc] = failures
+                delay = min(
+                    self.REFILL_BACKOFF_INITIAL
+                    * (2 ** min(failures - 1, 12)),
+                    self.REFILL_BACKOFF_MAX,
+                )
+                self._refill_after[dc] = time.monotonic() + delay
+                log.info("CF worker pool refill failed for DC%d, retry in %.0fs",
+                         dc, delay)
             log.debug("CF worker pool refilled DC%d: %d ready",
                       dc, len(bucket))
         finally:
             if self._refilling.get(dc) is asyncio.current_task():
                 self._refilling.pop(dc, None)
 
+    def _schedule_rotation(self, dc, fallback_dst, worker_domains):
+        if dc in self._rotating:
+            return
+        self._rotating[dc] = asyncio.create_task(
+            self._rotate(dc, fallback_dst, list(worker_domains)))
+
+    async def _rotate(self, dc, fallback_dst, worker_domains):
+        try:
+            while proxy_config.pool_size > 0:
+                bucket = self._idle.setdefault(dc, deque())
+                now = time.monotonic()
+                expired = []
+                ready = deque()
+                while bucket:
+                    ws, created, worker_domain = bucket.popleft()
+                    if self._is_stale(ws, created, now):
+                        expired.append(ws)
+                    else:
+                        ready.append((ws, created, worker_domain))
+                bucket.extend(ready)
+
+                if expired:
+                    for ws in expired:
+                        asyncio.create_task(self._quiet_close(ws))
+                    log.debug("CF worker pool rotated DC%d: %d stale, %d ready",
+                              dc, len(expired), len(bucket))
+
+                if len(bucket) < min(proxy_config.pool_size, self.PER_DC_LIMIT):
+                    self._schedule_refill(dc, fallback_dst, worker_domains)
+
+                wake_at = now + self.WS_POOL_CHECK_INTERVAL
+                if bucket:
+                    wake_at = min(wake_at, min(
+                        created + self.WS_POOL_MAX_AGE for _, created, _ in bucket))
+                if dc not in self._refilling and self._refill_after.get(dc, 0) > now:
+                    wake_at = min(wake_at, self._refill_after[dc])
+                await asyncio.sleep(max(0, wake_at - time.monotonic()))
+        finally:
+            if self._rotating.get(dc) is asyncio.current_task():
+                self._rotating.pop(dc, None)
+
     async def _connect_one(self, worker_domains, fallback_dst, dc):
         query = urlencode({
             'dst': fallback_dst,
@@ -309,6 +383,8 @@ class _CfWorkerPool:
                     secure=not proxy_config.disable_secure)
                 return ws, worker_domain
             except Exception as exc:
+                log.debug("CF worker pool connect DC%d via %s for %s failed: %r",
+                          dc, worker_domain, fallback_dst, exc)
                 self.report_failure(worker_domain, exc)
         return None
 
@@ -352,7 +428,8 @@ class _CfWorkerPool:
             if dc not in proxy_config.dc_redirects
         }
 
-        if not cf_fallbacks or not proxy_config.cfproxy_worker_domains:
+        if (proxy_config.pool_size <= 0 or not cf_fallbacks
+                or not proxy_config.cfproxy_worker_domains):
             return
 
         worker_domains = list(proxy_config.cfproxy_worker_domains)
@@ -363,7 +440,7 @@ class _CfWorkerPool:
 
     def reset(self):
         loop = asyncio.get_running_loop()
-        for task in self._refilling.values():
+        for task in list(self._rotating.values()) + list(self._refilling.values()):
             if not task.done() and task.get_loop() is loop:
                 task.cancel()
         for bucket in self._idle.values():
@@ -374,10 +451,13 @@ class _CfWorkerPool:
                     log.debug("CF worker pool close failed: %r", exc)
         self._idle.clear()
         self._refilling.clear()
+        self._rotating.clear()
+        self._refill_failures.clear()
+        self._refill_after.clear()
         self._exhausted_until.clear()
 
     async def close(self):
-        tasks = list(self._refilling.values())
+        tasks = list(self._rotating.values()) + list(self._refilling.values())
         self.reset()
         await asyncio.gather(*tasks, return_exceptions=True)
 

proxy/raw_websocket.py

@@ -225,8 +225,8 @@ class RawWebSocket:
         return None
 
     async def close(self):
-        if self._closed:
-            return
+        try:
+            if not self._closed:
                 self._closed = True
                 try:
                     self.writer.write(
@@ -234,6 +234,7 @@ class RawWebSocket:
                     await self.writer.drain()
                 except Exception:
                     pass
+        finally:
             try:
                 self.writer.close()
                 await self.writer.wait_closed()

proxy/tg_ws_proxy.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import os
 import sys
 import time
+import random
 import struct
 import asyncio
 import hashlib
@@ -24,7 +25,11 @@ from .stats import stats
 from .config import proxy_config, parse_dc_ip_list, start_cfproxy_domain_refresh, coerce_domain_list
 from .bridge import MsgSplitter, CryptoCtx, do_fallback, bridge_ws_reencrypt, reset_tcp_backoff
 from .raw_websocket import set_sock_opts
-from .fake_tls import proxy_to_masking_domain, verify_client_hello, build_server_hello, FakeTlsStream, TLS_RECORD_HANDSHAKE
+from .fake_tls import (
+    proxy_to_masking_domain, verify_client_hello, parse_client_hello, build_server_hello,
+    probe_appdata_len, set_profiled_appdata_len, FakeTlsStream, TLS_RECORD_HANDSHAKE,
+    SERVER_HELLO_DELAY,
+)
 from .balancer import balancer
 from .pool import ws_pool, cf_worker_pool
 from .cf_h2 import CfH2Pool
@@ -37,6 +42,8 @@ log = logging.getLogger('tg-mtproto-proxy')
 
 LISTENER_CHECK_INTERVAL = 5.0
 LISTENER_RESTART_DELAY = 1.0
+BAD_HANDSHAKE_DRAIN_TIMEOUT = 10.0
+BAD_HANDSHAKE_DRAIN_LIMIT = 64 * 1024
 cf_h2_pool: Optional[CfH2Pool] = None
 
 
@@ -162,7 +169,9 @@ async def _read_client_init(reader, writer, secret, label, masking):
         client_random, session_id, ts = tls_result
         log.debug("[%s] Fake TLS handshake ok (ts=%d)", label, ts)
 
-        server_hello = build_server_hello(secret, client_random, session_id)
+        server_hello = build_server_hello(
+            secret, client_random, session_id, parse_client_hello(client_hello))
+        await asyncio.sleep(random.uniform(*SERVER_HELLO_DELAY))
         writer.write(server_hello)
         await writer.drain()
 
@@ -244,6 +253,15 @@ def _build_crypto_ctx(client_dec_prekey_iv, secret, relay_init):
     return CryptoCtx(clt_decryptor, clt_encryptor, tg_encryptor, tg_decryptor)
 
 
+async def _drain_bad_handshake(reader):
+    remaining = BAD_HANDSHAKE_DRAIN_LIMIT
+    while remaining > 0:
+        chunk = await reader.read(min(4096, remaining))
+        if not chunk:
+            break
+        remaining -= len(chunk)
+
+
 async def _handle_client(reader, writer, secret: bytes):
     stats.connections_total += 1
     stats.connections_active += 1
@@ -265,8 +283,9 @@ async def _handle_client(reader, writer, secret: bytes):
             stats.connections_bad += 1
             log.warning("[%s] bad handshake (wrong secret or proto)", label)
             try:
-                while await clt_reader.read(4096):
-                    pass
+                await asyncio.wait_for(
+                    _drain_bad_handshake(clt_reader),
+                    timeout=BAD_HANDSHAKE_DRAIN_TIMEOUT)
             except Exception:
                 pass
             return
@@ -379,6 +398,19 @@ async def _run(stop_event: Optional[asyncio.Event] = None):
     secret_bytes = bytes.fromhex(proxy_config.secret)
     stopping = False
 
+    # Probe before listening so a domain that resolves to this proxy fails
+    # fast instead of answering itself
+    set_profiled_appdata_len(None)
+    if proxy_config.fake_tls_domain:
+        appdata_len = await probe_appdata_len(proxy_config.fake_tls_domain)
+        set_profiled_appdata_len(appdata_len)
+        if appdata_len is None:
+            log.warning("Fake TLS profile unavailable, ServerHello uses "
+                        "random record sizes")
+        else:
+            log.info("Fake TLS profile: first encrypted record %d bytes",
+                     appdata_len)
+
     def client_cb(r, w):
         if stopping:
             w.close()
@@ -675,7 +707,6 @@ def main():
             log_max_mb=args.log_max_mb,
             backups=args.log_backups,
         )
-        fh.setFormatter(log_fmt)
         fh.addFilter(DomainCensorFilter())
         root.addHandler(fh)
 

proxy/utils.py

@@ -143,12 +143,20 @@ class _PinnedHTTPSHandler(urllib.request.HTTPSHandler):
             return super().https_open(req)
 
 
+def set_x25519(context: ssl.SSLContext) -> None:
+    try:
+        context.set_ecdh_curve('X25519')
+    except Exception:
+        pass
+
+
 def create_ssl_context(*, check_hostname: bool = True) -> ssl.SSLContext:
     bundle_dir = getattr(sys, '_MEIPASS', None)
     cafile = (os.path.join(bundle_dir, 'certifi', 'cacert.pem')
               if bundle_dir else certifi.where())
     context = ssl.create_default_context(cafile=cafile)
     context.load_default_certs()
+    set_x25519(context)
     context.check_hostname = check_hostname
     return context
 

tests/test_cf_h2.py

@@ -812,7 +812,8 @@ class HttpMultiplexTest(unittest.IsolatedAsyncioTestCase):
                 self.assertEqual(len(older), 1)
                 self.assertEqual(len(latest), 4)
                 for first, second in zip(latest[1:], latest[2:]):
-                    self.assertGreaterEqual(second - first, .015)
+                    # Account for subtraction roundoff at the threshold.
+                    self.assertGreaterEqual(second - first, .015 - 1e-9)
                 self.assertFalse(channel.pending)
                 self.assertFalse(channel.replay_pending)
                 self.assertEqual(self.lane.inflight, 0)

tests/test_fake_tls.py

@@ -1,24 +1,111 @@
+import asyncio
+import datetime
 import hashlib
 import hmac
 import os
+import ssl
 import struct
+import tempfile
 import time
 import unittest
+from unittest.mock import AsyncMock, Mock, patch
 
+from proxy import fake_tls
 from proxy.fake_tls import (
+    APPDATA_MIN,
     CLIENT_RANDOM_LEN,
     CLIENT_RANDOM_OFFSET,
+    EXT_KEY_SHARE,
+    EXT_SUPPORTED_VERSIONS,
+    FALLBACK_APPDATA_RANGE,
+    GROUP_X25519,
+    GROUP_X25519MLKEM768,
+    MLKEM768_CIPHERTEXT_LEN,
     SESSION_ID_LEN,
     SESSION_ID_OFFSET,
     TLS_APPDATA_MAX,
+    TLS_RECORD_APPDATA,
+    TLS_RECORD_CCS,
     TLS_RECORD_HANDSHAKE,
+    X25519_KEY_LEN,
+    X25519MLKEM768_CLIENT_SHARE_LEN,
     build_server_hello,
+    first_appdata_record_len,
+    parse_client_hello,
+    probe_appdata_len,
+    set_profiled_appdata_len,
     verify_client_hello,
     wrap_tls_record,
 )
 
 SECRET = bytes.fromhex('00112233445566778899aabbccddeeff')
 
+# Telegram's current ClientHello offers these, like Chrome
+TELEGRAM_SUITES = (0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f)
+TELEGRAM_SHARES = ((GROUP_X25519MLKEM768, X25519MLKEM768_CLIENT_SHARE_LEN),
+                   (GROUP_X25519, X25519_KEY_LEN))
+
+
+def _sign_client_hello(body: bytearray, secret: bytes, timestamp: int) -> bytes:
+    digest = hmac.new(secret, bytes(body), hashlib.sha256).digest()
+    client_random = bytearray(digest[:CLIENT_RANDOM_LEN])
+    ts_bytes = struct.pack('<I', timestamp)
+    for i in range(4):
+        client_random[28 + i] = digest[28 + i] ^ ts_bytes[i]
+    body[CLIENT_RANDOM_OFFSET:CLIENT_RANDOM_OFFSET + CLIENT_RANDOM_LEN] = client_random
+    return bytes(body)
+
+
+def _structured_client_hello(suites=TELEGRAM_SUITES, shares=TELEGRAM_SHARES,
+                             secret: bytes = SECRET) -> bytes:
+    key_share_list = b''.join(struct.pack('>HH', group, length) + os.urandom(length)
+                              for group, length in shares)
+    extensions = (
+        struct.pack('>HH', 0x0000, 0)
+        + struct.pack('>HHH', EXT_KEY_SHARE, 2 + len(key_share_list),
+                      len(key_share_list))
+        + key_share_list
+        + struct.pack('>HHB', EXT_SUPPORTED_VERSIONS, 3, 2) + b'\x03\x04'
+    )
+    hello = (
+        b'\x03\x03' + b'\x00' * CLIENT_RANDOM_LEN
+        + b'\x20' + os.urandom(SESSION_ID_LEN)
+        + struct.pack('>H', 2 * len(suites))
+        + b''.join(struct.pack('>H', s) for s in suites)
+        + b'\x01\x00'
+        + struct.pack('>H', len(extensions)) + extensions
+    )
+    message = b'\x01' + struct.pack('>I', len(hello))[1:] + hello
+    body = bytearray(b'\x16\x03\x01' + struct.pack('>H', len(message)) + message)
+    return _sign_client_hello(body, secret, int(time.time()))
+
+
+def _split_records(data: bytes):
+    records = []
+    pos = 0
+    while pos < len(data):
+        rtype = data[pos]
+        length = struct.unpack_from('>H', data, pos + 3)[0]
+        records.append((rtype, data[pos + 5:pos + 5 + length]))
+        pos += 5 + length
+    return records
+
+
+def _server_hello_fields(record: bytes):
+    # record = handshake message: type(1) len(3) version(2) random(32) sid ...
+    pos = 4 + 2 + 32
+    pos += 1 + record[pos]
+    cipher = struct.unpack_from('>H', record, pos)[0]
+    pos += 3
+    ext_end = pos + 2 + struct.unpack_from('>H', record, pos)[0]
+    pos += 2
+    extensions = []
+    while pos < ext_end:
+        ext_type, ext_len = struct.unpack_from('>HH', record, pos)
+        extensions.append((ext_type, record[pos + 4:pos + 4 + ext_len]))
+        pos += 4 + ext_len
+    return cipher, extensions
+
 
 def _client_hello(secret: bytes = SECRET, timestamp: int = None,
                   session_id: bytes = None) -> bytes:
@@ -34,15 +121,7 @@ def _client_hello(secret: bytes = SECRET, timestamp: int = None,
     body[5] = 0x01
     body[43] = 0x20
     body[SESSION_ID_OFFSET:SESSION_ID_OFFSET + SESSION_ID_LEN] = session_id
-
-    digest = hmac.new(secret, bytes(body), hashlib.sha256).digest()
-    client_random = bytearray(digest[:CLIENT_RANDOM_LEN])
-    ts_bytes = struct.pack('<I', timestamp)
-    for i in range(4):
-        client_random[28 + i] = digest[28 + i] ^ ts_bytes[i]
-
-    body[CLIENT_RANDOM_OFFSET:CLIENT_RANDOM_OFFSET + CLIENT_RANDOM_LEN] = client_random
-    return bytes(body)
+    return _sign_client_hello(body, secret, timestamp)
 
 
 class VerifyClientHelloTest(unittest.TestCase):
@@ -80,7 +159,108 @@ class VerifyClientHelloTest(unittest.TestCase):
         self.assertIsNone(verify_client_hello(bytes(hello), SECRET))
 
 
+class ParseClientHelloTest(unittest.TestCase):
+    def test_extracts_suites_and_key_shares(self):
+        info = parse_client_hello(_structured_client_hello())
+        self.assertEqual(info.cipher_suites, frozenset(TELEGRAM_SUITES))
+        self.assertEqual(info.key_shares, dict(TELEGRAM_SHARES))
+
+    def test_structured_hello_still_verifies(self):
+        self.assertIsNotNone(verify_client_hello(_structured_client_hello(), SECRET))
+
+    def test_malformed_hello_yields_empty_info(self):
+        self.assertEqual(parse_client_hello(b'\x16\x03\x01\x00'),
+                         fake_tls._EMPTY_HELLO_INFO)
+        truncated = _structured_client_hello()[:60]
+        self.assertEqual(parse_client_hello(truncated), fake_tls._EMPTY_HELLO_INFO)
+
+    def test_unstructured_hello_does_not_raise(self):
+        parse_client_hello(_client_hello())
+
+
 class BuildServerHelloTest(unittest.TestCase):
+    def setUp(self):
+        set_profiled_appdata_len(None)
+
+    def tearDown(self):
+        set_profiled_appdata_len(None)
+
+    def _build(self, hello: bytes, client_random: bytes = None) -> bytes:
+        client_random = client_random or os.urandom(CLIENT_RANDOM_LEN)
+        return build_server_hello(SECRET, client_random, os.urandom(SESSION_ID_LEN),
+                                  parse_client_hello(hello))
+
+    def test_flight_is_server_hello_ccs_and_one_appdata_record(self):
+        records = _split_records(self._build(_structured_client_hello()))
+        self.assertEqual([r[0] for r in records],
+                         [TLS_RECORD_HANDSHAKE, TLS_RECORD_CCS, TLS_RECORD_APPDATA])
+        self.assertEqual(records[1][1], b'\x01')
+
+    def test_hmac_covers_whole_flight(self):
+        client_random = os.urandom(CLIENT_RANDOM_LEN)
+        response = self._build(_structured_client_hello(), client_random)
+        zeroed = bytearray(response)
+        zeroed[11:11 + 32] = b'\x00' * 32
+        expected = hmac.new(SECRET, client_random + bytes(zeroed),
+                            hashlib.sha256).digest()
+        self.assertEqual(response[11:11 + 32], expected)
+
+    def test_mirrors_post_quantum_group_when_offered(self):
+        records = _split_records(self._build(_structured_client_hello()))
+        cipher, extensions = _server_hello_fields(records[0][1])
+        self.assertEqual([e[0] for e in extensions],
+                         [EXT_SUPPORTED_VERSIONS, EXT_KEY_SHARE])
+        key_share = extensions[1][1]
+        group, length = struct.unpack_from('>HH', key_share)
+        self.assertEqual(group, GROUP_X25519MLKEM768)
+        self.assertEqual(length, MLKEM768_CIPHERTEXT_LEN + X25519_KEY_LEN)
+        self.assertEqual(len(key_share), 4 + length)
+        self.assertEqual(cipher, 0x1301)
+
+    def test_falls_back_to_x25519(self):
+        hello = _structured_client_hello(shares=((GROUP_X25519, X25519_KEY_LEN),))
+        records = _split_records(self._build(hello))
+        _, extensions = _server_hello_fields(records[0][1])
+        group, length = struct.unpack_from('>HH', extensions[1][1])
+        self.assertEqual((group, length), (GROUP_X25519, X25519_KEY_LEN))
+        # real X25519 public keys are below 2^255
+        self.assertEqual(extensions[1][1][-1] & 0x80, 0)
+
+    def test_ignores_post_quantum_share_of_wrong_size(self):
+        hello = _structured_client_hello(
+            shares=((GROUP_X25519MLKEM768, 100), (GROUP_X25519, X25519_KEY_LEN)))
+        records = _split_records(self._build(hello))
+        _, extensions = _server_hello_fields(records[0][1])
+        self.assertEqual(struct.unpack_from('>H', extensions[1][1])[0], GROUP_X25519)
+
+    def test_picks_cipher_offered_by_client(self):
+        hello = _structured_client_hello(suites=(0x1302, 0x1303))
+        cipher, _ = _server_hello_fields(_split_records(self._build(hello))[0][1])
+        self.assertEqual(cipher, 0x1303)
+
+    def test_telegram_flight_matches_telemt_size(self):
+        # telemt answered every Telegram client with exactly 1290 bytes when the
+        # masking site's first encrypted record was below the 64-byte floor
+        set_profiled_appdata_len(30)
+        self.assertEqual(len(self._build(_structured_client_hello())), 1290)
+
+    def test_uses_profiled_appdata_len(self):
+        set_profiled_appdata_len(3000)
+        records = _split_records(self._build(_structured_client_hello()))
+        self.assertEqual(len(records[2][1]), 3000)
+
+    def test_profiled_len_is_clamped(self):
+        set_profiled_appdata_len(10)
+        self.assertEqual(fake_tls._profiled_appdata_len, APPDATA_MIN)
+        set_profiled_appdata_len(10 ** 6)
+        self.assertEqual(fake_tls._profiled_appdata_len, TLS_APPDATA_MAX)
+
+    def test_fallback_len_stays_in_range(self):
+        for _ in range(20):
+            records = _split_records(self._build(_structured_client_hello()))
+            self.assertGreaterEqual(len(records[2][1]), FALLBACK_APPDATA_RANGE[0])
+            self.assertLessEqual(len(records[2][1]), FALLBACK_APPDATA_RANGE[1])
+
     def test_echoes_session_id_and_binds_client_random(self):
         session_id = os.urandom(SESSION_ID_LEN)
         client_random = os.urandom(CLIENT_RANDOM_LEN)
@@ -106,6 +286,119 @@ class BuildServerHelloTest(unittest.TestCase):
         self.assertGreater(len(sizes), 1)
 
 
+def _record(rtype: int, length: int) -> bytes:
+    return bytes([rtype]) + b'\x03\x03' + struct.pack('>H', length) + b'\x00' * length
+
+
+class FirstAppdataRecordLenTest(unittest.TestCase):
+    def test_finds_record_after_server_hello_and_ccs(self):
+        flight = (_record(TLS_RECORD_HANDSHAKE, 122) + _record(TLS_RECORD_CCS, 1)
+                  + _record(TLS_RECORD_APPDATA, 27) + _record(TLS_RECORD_APPDATA, 3000))
+        self.assertEqual(first_appdata_record_len(flight), 27)
+
+    def test_needs_more_data_until_header_arrives(self):
+        partial = _record(TLS_RECORD_HANDSHAKE, 122) + _record(TLS_RECORD_CCS, 1)
+        self.assertIsNone(first_appdata_record_len(partial))
+        self.assertIsNone(first_appdata_record_len(partial + b'\x17\x03'))
+
+    def test_alert_means_no_profile(self):
+        self.assertEqual(first_appdata_record_len(_record(0x15, 2)), 0)
+
+
+def _self_signed_context() -> ssl.SSLContext:
+    from cryptography import x509
+    from cryptography.hazmat.primitives import hashes, serialization
+    from cryptography.hazmat.primitives.asymmetric import ec
+    from cryptography.x509.oid import NameOID
+
+    key = ec.generate_private_key(ec.SECP256R1())
+    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'localhost')])
+    now = datetime.datetime.now(datetime.timezone.utc)
+    cert = (x509.CertificateBuilder()
+            .subject_name(name).issuer_name(name)
+            .public_key(key.public_key())
+            .serial_number(x509.random_serial_number())
+            .not_valid_before(now - datetime.timedelta(days=1))
+            .not_valid_after(now + datetime.timedelta(days=1))
+            .sign(key, hashes.SHA256()))
+
+    with tempfile.TemporaryDirectory() as tmp:
+        cert_path = os.path.join(tmp, 'cert.pem')
+        key_path = os.path.join(tmp, 'key.pem')
+        with open(cert_path, 'wb') as f:
+            f.write(cert.public_bytes(serialization.Encoding.PEM))
+        with open(key_path, 'wb') as f:
+            f.write(key.private_bytes(serialization.Encoding.PEM,
+                                      serialization.PrivateFormat.PKCS8,
+                                      serialization.NoEncryption()))
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        ctx.load_cert_chain(cert_path, key_path)
+    return ctx
+
+
+@unittest.skipUnless(getattr(ssl, 'HAS_TLSv1_3', False), 'needs TLS 1.3 in ssl')
+class ProbeAppdataLenTest(unittest.TestCase):
+    def test_continuous_records_do_not_extend_probe_timeout(self):
+        async def scenario():
+            async def read(_size):
+                await asyncio.sleep(0.01)
+                return _record(TLS_RECORD_CCS, 1)
+
+            reader = Mock(read=AsyncMock(side_effect=read))
+            writer = Mock(drain=AsyncMock())
+            with patch.object(fake_tls.asyncio, 'open_connection',
+                              AsyncMock(return_value=(reader, writer))):
+                result = await asyncio.wait_for(
+                    probe_appdata_len('localhost', timeout=0.1), timeout=2)
+            self.assertIsNone(result)
+            self.assertGreater(reader.read.await_count, 1)
+            writer.close.assert_called_once()
+
+        asyncio.run(scenario())
+
+    def test_blocked_drain_obeys_probe_timeout(self):
+        async def scenario():
+            reader = Mock(read=AsyncMock())
+            writer = Mock(drain=AsyncMock(side_effect=asyncio.Event().wait))
+            with patch.object(fake_tls.asyncio, 'open_connection',
+                              AsyncMock(return_value=(reader, writer))):
+                result = await asyncio.wait_for(
+                    probe_appdata_len('localhost', timeout=0.1), timeout=2)
+            self.assertIsNone(result)
+            reader.read.assert_not_called()
+            hello = writer.write.call_args[0][0]
+            self.assertEqual(parse_client_hello(hello).key_shares,
+                             {GROUP_X25519: X25519_KEY_LEN})
+            writer.close.assert_called_once()
+
+        asyncio.run(scenario())
+
+    def test_measures_first_encrypted_record_of_real_tls13_server(self):
+        async def scenario():
+            server = await asyncio.start_server(
+                lambda r, w: None, '127.0.0.1', 0, ssl=_self_signed_context())
+            port = server.sockets[0].getsockname()[1]
+            try:
+                return await probe_appdata_len('127.0.0.1', port, timeout=3)
+            finally:
+                server.close()
+                await server.wait_closed()
+
+        length = asyncio.run(scenario())
+        self.assertIsNotNone(length)
+        self.assertGreater(length, 0)
+
+    def test_unreachable_site_gives_no_profile(self):
+        async def scenario():
+            server = await asyncio.start_server(lambda r, w: None, '127.0.0.1', 0)
+            port = server.sockets[0].getsockname()[1]
+            server.close()
+            await server.wait_closed()
+            return await probe_appdata_len('127.0.0.1', port, timeout=1)
+
+        self.assertIsNone(asyncio.run(scenario()))
+
+
 class WrapTlsRecordTest(unittest.TestCase):
     def test_short_payload_becomes_one_record(self):
         wrapped = wrap_tls_record(b'hello')

tests/test_pool.py

@@ -9,7 +9,7 @@ from unittest import mock
 from proxy.config import proxy_config
 from proxy.pool import _CfWorkerPool, _WsPool
 from proxy.raw_websocket import RawWebSocket, WsHandshakeError
-from proxy.utils import ws_domains
+from proxy.utils import DC_DEFAULT_IPS, ws_domains
 
 
 class _StopRotation(Exception):
@@ -242,6 +242,154 @@ class WsPoolTest(unittest.IsolatedAsyncioTestCase):
 
 
 class CfWorkerPoolTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.pool = _CfWorkerPool()
+        self.pool.WS_POOL_CHECK_INTERVAL = .01
+        for name, value in [
+                ('pool_size', 4),
+                ('dc_redirects', {dc: ip for dc, ip in DC_DEFAULT_IPS.items() if dc != 2}),
+                ('cfproxy_worker_domains', ['worker.example'])]:
+            patcher = mock.patch.object(proxy_config, name, value)
+            patcher.start()
+            self.addCleanup(patcher.stop)
+        self.addAsyncCleanup(self.pool.close)
+
+    async def wait_for(self, predicate):
+        async def wait():
+            while not predicate():
+                await asyncio.sleep(.001)
+        await asyncio.wait_for(wait(), 1)
+
+    async def test_failed_warmup_recovers_without_new_client_requests(self):
+        ready = _open_ws()
+        self.pool.REFILL_BACKOFF_INITIAL = .03
+        with mock.patch.object(self.pool, '_connect_one',
+                               side_effect=[None, (ready, 'worker.example')]) as dial:
+            await self.pool.warmup()
+            await self.wait_for(lambda: 2 in self.pool._refill_after)
+            self.assertFalse(self.pool._refilling)
+            await self.wait_for(lambda: bool(self.pool._idle.get(2)))
+            self.assertEqual(dial.await_count, 2)
+            self.assertIs(self.pool._idle[2][0][0], ready)
+            self.assertNotIn(2, self.pool._refill_failures)
+            self.assertNotIn(2, self.pool._refill_after)
+
+    async def test_stale_connections_are_replaced(self):
+        for state in ('expired', 'eof', 'exception', 'closed', 'closing'):
+            with self.subTest(state=state):
+                stale, ready = _open_ws(), _open_ws()
+                created = time.monotonic()
+                if state == 'expired':
+                    created -= self.pool.WS_POOL_MAX_AGE
+                elif state == 'eof':
+                    stale.reader.feed_eof()
+                elif state == 'exception':
+                    stale.reader.set_exception(ConnectionResetError())
+                elif state == 'closed':
+                    stale._closed = True
+                else:
+                    stale.writer.transport.is_closing = lambda: True
+                self.pool._idle[2] = deque([(stale, created, 'worker.example')])
+                with mock.patch.object(self.pool, '_connect_one',
+                                       return_value=(ready, 'worker.example')):
+                    self.assertIsNone(await self.pool.get(2, '192.0.2.2', ['worker.example']))
+                    await self.wait_for(lambda: bool(self.pool._idle[2]))
+                    self.assertIs(self.pool._idle[2][0][0], ready)
+                    stale.writer.close.assert_called_once()
+                    await self.pool.close()
+
+    async def test_rotation_replaces_idle_connection_without_client_requests(self):
+        stale, ready = _open_ws(), _open_ws()
+        self.pool.WS_POOL_MAX_AGE = .05
+        self.pool._idle[2] = deque([(stale, time.monotonic(), 'worker.example')])
+        with mock.patch.object(self.pool, '_connect_one',
+                               return_value=(ready, 'worker.example')) as dial:
+            await self.pool.warmup()
+            await self.wait_for(lambda: stale.writer.close.called)
+            self.assertIs(self.pool._idle[2][0][0], ready)
+            stale.writer.close.assert_called_once()
+            dial.assert_awaited_once()
+
+    async def test_concurrent_misses_start_only_one_background_connection(self):
+        started = asyncio.Event()
+
+        async def connect(*args):
+            started.set()
+            await asyncio.Future()
+
+        with mock.patch.object(self.pool, '_connect_one', side_effect=connect) as dial:
+            for _ in range(10):
+                self.assertIsNone(await self.pool.get(2, '192.0.2.2', ['worker.example']))
+            await asyncio.wait_for(started.wait(), 1)
+            self.assertEqual(dial.await_count, 1)
+
+    async def test_failed_refills_use_capped_exponential_backoff(self):
+        ready = _open_ws()
+        self.pool.REFILL_BACKOFF_INITIAL = 1
+        self.pool.REFILL_BACKOFF_MAX = 2
+        with mock.patch.object(self.pool, '_connect_one',
+                               side_effect=[None, None, None, (ready, 'worker.example')]), \
+                mock.patch.object(self.pool, '_schedule_rotation'):
+            for failures, delay in enumerate((1, 2, 2), 1):
+                self.pool._refill_after.pop(2, None)
+                started = time.monotonic()
+                self.pool._schedule_refill(2, '192.0.2.2', ['worker.example'])
+                await asyncio.gather(*self.pool._refilling.values())
+                self.assertFalse(self.pool._refilling)
+                self.assertEqual(self.pool._refill_failures[2], failures)
+                self.assertGreaterEqual(self.pool._refill_after[2], started + delay)
+                self.assertLessEqual(self.pool._refill_after[2], time.monotonic() + delay)
+            self.pool._refill_after.pop(2)
+            self.pool._schedule_refill(2, '192.0.2.2', ['worker.example'])
+            await asyncio.gather(*self.pool._refilling.values())
+        self.assertIs(self.pool._idle[2][0][0], ready)
+        self.assertNotIn(2, self.pool._refill_failures)
+        self.assertNotIn(2, self.pool._refill_after)
+
+    async def test_misses_during_backoff_do_not_retry_and_shutdown_cancels_retry(self):
+        self.pool.REFILL_BACKOFF_INITIAL = 60
+        with mock.patch.object(self.pool, '_connect_one', return_value=None) as dial:
+            await self.pool.warmup()
+            await self.wait_for(lambda: 2 in self.pool._refill_after)
+            for _ in range(10):
+                self.assertIsNone(await self.pool.get(2, '192.0.2.2', ['worker.example']))
+            await asyncio.sleep(.01)
+            self.assertEqual(dial.await_count, 1)
+            self.assertFalse(self.pool._refilling)
+            self.assertTrue(self.pool._rotating)
+            tasks = list(self.pool._rotating.values())
+            await self.pool.close()
+            self.assertFalse(self.pool._refilling)
+            self.assertFalse(self.pool._rotating)
+            self.assertFalse(self.pool._refill_after)
+            self.assertFalse(self.pool._refill_failures)
+            self.assertTrue(all(task.done() for task in tasks))
+
+    async def test_hit_does_not_reset_failed_refill_backoff(self):
+        ready = _open_ws()
+        self.pool._idle[2] = deque([(ready, time.monotonic(), 'worker.example')])
+        self.pool._refill_failures[2] = 3
+        retry_at = time.monotonic() + 60
+        self.pool._refill_after[2] = retry_at
+        with mock.patch.object(self.pool, '_schedule_rotation'):
+            self.assertEqual(await self.pool.get(2, '192.0.2.2', ['worker.example']),
+                             (ready, 'worker.example'))
+        self.assertEqual(self.pool._refill_failures[2], 3)
+        self.assertEqual(self.pool._refill_after[2], retry_at)
+        self.assertFalse(self.pool._refilling)
+
+    async def test_zero_size_and_missing_worker_domains_do_not_start_refills(self):
+        with mock.patch.object(self.pool, '_connect_one') as dial:
+            with mock.patch.object(proxy_config, 'pool_size', 0):
+                await self.pool.warmup()
+                self.assertIsNone(await self.pool.get(2, '192.0.2.2', ['worker.example']))
+                self.assertFalse(self.pool._refilling)
+                self.assertFalse(self.pool._rotating)
+            self.assertIsNone(await self.pool.get(2, '192.0.2.2', []))
+            self.assertFalse(self.pool._refilling)
+            self.assertFalse(self.pool._rotating)
+            dial.assert_not_called()
+
     async def test_close_cancels_refill_and_closes_unclaimed_connections(self):
         pool = _CfWorkerPool()
         ready = _open_ws()
@@ -263,6 +411,9 @@ class CfWorkerPoolTest(unittest.IsolatedAsyncioTestCase):
             await pool.close()
         self.assertFalse(pool._idle)
         self.assertFalse(pool._refilling)
+        self.assertFalse(pool._rotating)
+        self.assertFalse(pool._refill_after)
+        self.assertFalse(pool._refill_failures)
         ready.writer.close.assert_called_once()
         idle.writer.close.assert_called_once()
 

tests/test_raw_websocket.py

@@ -143,6 +143,53 @@ class ParseCloseTest(unittest.TestCase):
         self.assertEqual(RawWebSocket._parse_close(b''), (None, ''))
 
 
+class CloseTest(unittest.IsolatedAsyncioTestCase):
+    async def test_peer_close_still_closes_transport_without_another_close_frame(self):
+        reader = asyncio.StreamReader()
+        reader.feed_data(_raw_frame(RawWebSocket.OP_CLOSE, b'\x03\xe8'))
+        writer = Mock(drain=AsyncMock(), wait_closed=AsyncMock())
+        ws = RawWebSocket(reader, writer)
+
+        self.assertIsNone(await ws.recv())
+        await ws.close()
+
+        writer.close.assert_called_once()
+        writer.wait_closed.assert_awaited_once()
+        self.assertEqual(writer.write.call_count, 1)
+
+    async def test_repeated_close_sends_only_one_close_frame(self):
+        writer = Mock(drain=AsyncMock(), wait_closed=AsyncMock())
+        ws = RawWebSocket(asyncio.StreamReader(), writer)
+
+        await ws.close()
+        await ws.close()
+
+        self.assertTrue(ws._closed)
+        self.assertEqual(writer.write.call_count, 1)
+        writer.close.assert_called()
+
+    async def test_cancelling_close_during_drain_still_closes_transport(self):
+        draining = asyncio.Event()
+
+        async def drain():
+            draining.set()
+            await asyncio.Future()
+
+        writer = Mock(drain=AsyncMock(side_effect=drain), wait_closed=AsyncMock())
+        ws = RawWebSocket(asyncio.StreamReader(), writer)
+        task = asyncio.create_task(ws.close())
+        try:
+            await asyncio.wait_for(draining.wait(), 1)
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await task
+            writer.close.assert_called_once()
+            writer.wait_closed.assert_awaited_once()
+        finally:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+
+
 class HandshakeErrorTest(unittest.TestCase):
     def test_redirect_status_codes(self):
         for code in (301, 302, 303, 307, 308):

utils/logging_setup.py

@@ -1,4 +1,4 @@
-"""Shared construction of the rotating log file handler.
+"""Shared construction and formatting of the rotating log file handler.
 
 Centralizes the rotation invariant so both the tray and the CLI log paths
 behave identically and the file can never grow without bound (issue #885).
@@ -31,9 +31,14 @@ def build_log_handler(
     """
     max_bytes = max(_MIN_BYTES, int(log_max_mb * 1024 * 1024))
     backup_count = max(_MIN_BACKUPS, int(backups))
-    return logging.handlers.RotatingFileHandler(
+    handler = logging.handlers.RotatingFileHandler(
         path,
         maxBytes=max_bytes,
         backupCount=backup_count,
         encoding="utf-8",
     )
+    handler.setFormatter(logging.Formatter(
+        "%(asctime)s  %(levelname)-5s  %(name)s  %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    ))
+    return handler

utils/tray_common.py

@@ -225,7 +225,6 @@ def save_config(cfg: dict) -> None:
 
 # logging
 
-_LOG_FMT_FILE = "%(asctime)s  %(levelname)-5s  %(name)s  %(message)s"
 _LOG_FMT_CONSOLE = "%(asctime)s  %(levelname)-5s  %(message)s"
 
 
@@ -238,7 +237,6 @@ def setup_logging(verbose: bool = False, log_max_mb: float = 5) -> None:
 
     fh = build_log_handler(str(LOG_FILE), log_max_mb=log_max_mb, backups=1)
     fh.setLevel(logging.DEBUG)
-    fh.setFormatter(logging.Formatter(_LOG_FMT_FILE, datefmt="%Y-%m-%d %H:%M:%S"))
     fh.addFilter(DomainCensorFilter())
     root.addHandler(fh)