From 7fb01a7d0515155550f3df801c539480c2fe94bd Mon Sep 17 00:00:00 2001 From: Denis <7009699+someden@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:25:21 +0100 Subject: [PATCH] Fake TLS: mirror client key share and profile the masking site (#1473) --- proxy/fake_tls.py | 253 +++++++++++++++++++++++++++++++++---- proxy/tg_ws_proxy.py | 24 +++- tests/test_fake_tls.py | 275 +++++++++++++++++++++++++++++++++++++++-- 3 files changed, 515 insertions(+), 37 deletions(-) diff --git a/proxy/fake_tls.py b/proxy/fake_tls.py index ea6379d..41abcca 100644 --- a/proxy/fake_tls.py +++ b/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: - 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 +232,84 @@ 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.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() + + reader, writer = await asyncio.wait_for( + asyncio.open_connection(domain, port), timeout=timeout) + writer.write(client_hello) + await writer.drain() + + flight = bytearray() + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while True: + chunk = await asyncio.wait_for( + reader.read(16384), timeout=max(0.1, deadline - loop.time())) + 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 diff --git a/proxy/tg_ws_proxy.py b/proxy/tg_ws_proxy.py index d81baf7..92cd961 100644 --- a/proxy/tg_ws_proxy.py +++ b/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() diff --git a/tests/test_fake_tls.py b/tests/test_fake_tls.py index 168246a..f6e3955 100644 --- a/tests/test_fake_tls.py +++ b/tests/test_fake_tls.py @@ -1,24 +1,110 @@ +import asyncio +import datetime import hashlib import hmac import os +import ssl import struct +import tempfile import time import unittest +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(' 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 +120,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(' 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 +285,84 @@ 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_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')