Version bump

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,5 @@ 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

@@ -35,13 +35,14 @@ 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                           |
+| Variable                | Description                                        | Default                               |
-| ----------------------- | -------------------------------- | --------------------------------- |
+| ----------------------- | -------------------------------------------------- | ------------------------------------- |
-| `TG_WS_PROXY_HOST`      | Address for incoming connections | `0.0.0.0`                        |
+| `TG_WS_PROXY_HOST`      | Address for incoming connections                   | `0.0.0.0`                             |
-| `TG_WS_PROXY_PORT`      | Port inside the container        | `1443`                           |
+| `TG_WS_PROXY_PORT`      | Port inside the container                          | `1443`                                |
-| `TG_WS_PROXY_SECRET`    | Secret key                       | `random`                         |
+| `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_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_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

@@ -35,13 +35,14 @@ tg://proxy?server=172.17.0.2&port=1443&secret=dd68f127db1d...
 
 Все настройки задаются переменными окружения при запуске контейнера:
 
-| Переменная              | Описание                          | По умолчанию                          |
+| Переменная              | Описание                                                      | По умолчанию                          |
-| ----------------------- | --------------------------------- | ------------------------------------- |
+| ----------------------- | ------------------------------------------------------------- | ------------------------------------- |
-| `TG_WS_PROXY_HOST`      | `Адрес для приёма подключений`    | `0.0.0.0`                             |
+| `TG_WS_PROXY_HOST`      | `Адрес для приёма подключений`                                | `0.0.0.0`                             |
-| `TG_WS_PROXY_PORT`      | `Порт внутри контейнера`          | `1443`                                |
+| `TG_WS_PROXY_PORT`      | `Порт внутри контейнера`                                      | `1443`                                |
-| `TG_WS_PROXY_SECRET`    | `Секретный ключ`                  | `random`                              |
+| `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_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_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),
+    filevers=(1, 11, 3, 0),
-    prodvers=(1, 11, 2, 0),
+    prodvers=(1, 11, 3, 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.3.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.3.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.3"
 
 __all__ = ["__version__", "get_link_host", "proxy_config", "parse_dc_ip_list", "build_github_opener", "coerce_domain_list"]

proxy/fake_tls.py

@@ -5,13 +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
 
+try:
+    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+except ImportError:
+    X25519PrivateKey = None
+
 
 log = logging.getLogger('tg-mtproto-proxy')
 
@@ -32,26 +38,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 +114,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:
+def parse_client_hello(data: bytes) -> ClientHelloInfo:
-    sh = bytearray(_SERVER_HELLO_TEMPLATE)
+    """Extract offered cipher suites and key_share groups from a ClientHello.
-    sh[_SH_SESSID_OFF:_SH_SESSID_OFF + 32] = session_id
-    sh[_SH_PUBKEY_OFF:_SH_PUBKEY_OFF + 32] = os.urandom(32)
 
-    ccs = _CCS_FRAME
+    Returns empty info on malformed input so the caller falls back to the
-    encrypted_size = random.randint(1900, 2100)
+    plain X25519 / TLS_AES_128_GCM_SHA256 answer.
-    encrypted_data = os.urandom(encrypted_size)
+    """
-    app_record = b'\x17\x03\x03' + struct.pack('>H', encrypted_size) + encrypted_data
+    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 +232,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()
+        ctx.set_ecdh_curve('X25519')
+        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/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
@@ -162,7 +167,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()
 
@@ -379,6 +386,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()

proxy/utils.py

@@ -149,6 +149,7 @@ def create_ssl_context(*, check_hostname: bool = True) -> ssl.SSLContext:
               if bundle_dir else certifi.where())
     context = ssl.create_default_context(cafile=cafile)
     context.load_default_certs()
+    context.set_ecdh_curve('X25519')
     context.check_hostname = check_hostname
     return context
 

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
-
+    return _sign_client_hello(body, secret, timestamp)
-    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)
 
 
 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')