Upload 7 files

README.md

@@ -1,10 +1,39 @@
----
-title: MT
-emoji: 🐢
-colorFrom: purple
-colorTo: green
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# MTProto Proxy with WebUI
+
+This is a fast, lightweight MTProto Proxy (based on the excellent async Python implementation by `alexbers`), wrapped in a single Docker container along with a real-time Flask WebUI dashboard to monitor users and system metrics.
+
+## Features
+- **High Performance**: Uses Python 3.12 with `uvloop` for maximum async performance.
+- **WebUI Dashboard**: Live monitoring of CPU load, Memory usage, and Active MTProto Connections on port `7860`.
+- **All-in-One**: Single Docker container for both the proxy and the web interface.
+- **Secure**: Supports MTProto TLS mode for bypassing DPI.
+
+## Getting Started
+
+### 1. Configure the Proxy
+Open `config.py` and modify it as needed. By default, it runs on port `443` and uses TLS mode.
+Add or modify users in the `USERS` dictionary. Generate a secure secret (32 hex characters) for each user:
+```python
+USERS = {
+    "user1": "b08b584d41348b61a357549641743a29", # Replace with your random secret
+}
+```
+
+### 2. Build the Docker Image
+```bash
+docker build -t mtproto-proxy-webui .
+```
+
+### 3. Run the Docker Container
+Run the container, making sure to map the MTProto proxy port (default `443`) and the WebUI port (`7860`):
+```bash
+docker run -d --name mtproto-webui -p 443:443 -p 7860:7860 mtproto-proxy-webui
+```
+
+*(Note: If you change the `PORT` in `config.py`, make sure to update the `-p` mapping accordingly, e.g., `-p 8443:8443`)*
+
+## Accessing the WebUI
+Once the container is running, open your browser and navigate to:
+`http://<your-server-ip>:7860`
+
+You will see a real-time dashboard displaying system metrics and active connection statistics for each user.

config.py

@@ -0,0 +1,26 @@
+PORT = 443
+
+# name -> secret (32 hex chars)
+USERS = {
+    "tg":  "00000000000000000000000000000001",
+}
+
+MODES = {
+    # Classic mode, easy to detect
+    "classic": False,
+
+    # Makes the proxy harder to detect
+    # Can be incompatible with very old clients
+    "secure": False,
+
+    # Makes the proxy even more hard to detect
+    # Can be incompatible with old clients
+    "tls": True
+}
+
+# The domain for TLS mode, bad clients are proxied there
+# Use random existing domain, proxy checks it on start
+TLS_DOMAIN = "www.google.com"
+
+# Tag for advertising, obtainable from @MTProxybot
+AD_TAG = ""

mtprotoproxy.py

@@ -0,0 +1,2414 @@
+#!/usr/bin/env python3
+
+import asyncio
+import socket
+import urllib.parse
+import urllib.request
+import collections
+import time
+import datetime
+import hmac
+import base64
+import hashlib
+import random
+import binascii
+import sys
+import re
+import runpy
+import signal
+import os
+import stat
+import traceback
+
+
+TG_DATACENTER_PORT = 443
+
+TG_DATACENTERS_V4 = [
+    "149.154.175.50", "149.154.167.51", "149.154.175.100",
+    "149.154.167.91", "149.154.171.5"
+]
+
+TG_DATACENTERS_V6 = [
+    "2001:b28:f23d:f001::a", "2001:67c:04e8:f002::a", "2001:b28:f23d:f003::a",
+    "2001:67c:04e8:f004::a", "2001:b28:f23f:f005::a"
+]
+
+# This list will be updated in the runtime
+TG_MIDDLE_PROXIES_V4 = {
+    1: [("149.154.175.50", 8888)], -1: [("149.154.175.50", 8888)],
+    2: [("149.154.161.144", 8888)], -2: [("149.154.161.144", 8888)],
+    3: [("149.154.175.100", 8888)], -3: [("149.154.175.100", 8888)],
+    4: [("91.108.4.136", 8888)], -4: [("149.154.165.109", 8888)],
+    5: [("91.108.56.183", 8888)], -5: [("91.108.56.183", 8888)]
+}
+
+TG_MIDDLE_PROXIES_V6 = {
+    1: [("2001:b28:f23d:f001::d", 8888)], -1: [("2001:b28:f23d:f001::d", 8888)],
+    2: [("2001:67c:04e8:f002::d", 80)], -2: [("2001:67c:04e8:f002::d", 80)],
+    3: [("2001:b28:f23d:f003::d", 8888)], -3: [("2001:b28:f23d:f003::d", 8888)],
+    4: [("2001:67c:04e8:f004::d", 8888)], -4: [("2001:67c:04e8:f004::d", 8888)],
+    5: [("2001:b28:f23f:f005::d", 8888)], -5: [("2001:b28:f23f:f005::d", 8888)]
+}
+
+PROXY_SECRET = bytes.fromhex(
+    "c4f9faca9678e6bb48ad6c7e2ce5c0d24430645d554addeb55419e034da62721" +
+    "d046eaab6e52ab14a95a443ecfb3463e79a05a66612adf9caeda8be9a80da698" +
+    "6fb0a6ff387af84d88ef3a6413713e5c3377f6e1a3d47d99f5e0c56eece8f05c" +
+    "54c490b079e31bef82ff0ee8f2b0a32756d249c5f21269816cb7061b265db212"
+)
+
+SKIP_LEN = 8
+PREKEY_LEN = 32
+KEY_LEN = 32
+IV_LEN = 16
+HANDSHAKE_LEN = 64
+PROTO_TAG_POS = 56
+DC_IDX_POS = 60
+
+MIN_CERT_LEN = 1024
+
+PROTO_TAG_ABRIDGED = b"\xef\xef\xef\xef"
+PROTO_TAG_INTERMEDIATE = b"\xee\xee\xee\xee"
+PROTO_TAG_SECURE = b"\xdd\xdd\xdd\xdd"
+
+CBC_PADDING = 16
+PADDING_FILLER = b"\x04\x00\x00\x00"
+
+MIN_MSG_LEN = 12
+MAX_MSG_LEN = 2 ** 24
+
+STAT_DURATION_BUCKETS = [0.1, 0.5, 1, 2, 5, 15, 60, 300, 600, 1800, 2**31 - 1]
+
+my_ip_info = {"ipv4": None, "ipv6": None}
+used_handshakes = collections.OrderedDict()
+client_ips = collections.OrderedDict()
+last_client_ips = {}
+disable_middle_proxy = False
+is_time_skewed = False
+fake_cert_len = random.randrange(1024, 4096)
+mask_host_cached_ip = None
+last_clients_with_time_skew = {}
+last_clients_with_same_handshake = collections.Counter()
+proxy_start_time = 0
+proxy_links = []
+
+stats = collections.Counter()
+user_stats = collections.defaultdict(collections.Counter)
+
+config = {}
+
+
+def init_config():
+    global config
+    # we use conf_dict to protect the original config from exceptions when reloading
+    if len(sys.argv) < 2:
+        conf_dict = runpy.run_module("config")
+    elif len(sys.argv) == 2:
+        # launch with own config
+        conf_dict = runpy.run_path(sys.argv[1])
+    else:
+        # undocumented way of launching
+        conf_dict = {}
+        conf_dict["PORT"] = int(sys.argv[1])
+        secrets = sys.argv[2].split(",")
+        conf_dict["USERS"] = {"user%d" % i: secrets[i].zfill(32) for i in range(len(secrets))}
+        conf_dict["MODES"] = {"classic": False, "secure": True, "tls": True}
+        if len(sys.argv) > 3:
+            conf_dict["AD_TAG"] = sys.argv[3]
+        if len(sys.argv) > 4:
+            conf_dict["TLS_DOMAIN"] = sys.argv[4]
+            conf_dict["MODES"] = {"classic": False, "secure": False, "tls": True}
+
+    conf_dict = {k: v for k, v in conf_dict.items() if k.isupper()}
+
+    conf_dict.setdefault("PORT", 3256)
+    conf_dict.setdefault("USERS", {"tg":  "00000000000000000000000000000000"})
+    conf_dict["AD_TAG"] = bytes.fromhex(conf_dict.get("AD_TAG", ""))
+
+    for user, secret in conf_dict["USERS"].items():
+        if not re.fullmatch("[0-9a-fA-F]{32}", secret):
+            fixed_secret = re.sub(r"[^0-9a-fA-F]", "", secret).zfill(32)[:32]
+
+            print_err("Bad secret for user %s, should be 32 hex chars, got %s. " % (user, secret))
+            print_err("Changing it to %s" % fixed_secret)
+
+            conf_dict["USERS"][user] = fixed_secret
+
+    # load advanced settings
+
+    # use middle proxy, necessary to show ad
+    conf_dict.setdefault("USE_MIDDLE_PROXY", len(conf_dict["AD_TAG"]) == 16)
+
+    # if IPv6 available, use it by default, IPv6 with middle proxies is unstable now
+    conf_dict.setdefault("PREFER_IPV6", socket.has_ipv6 and not conf_dict["USE_MIDDLE_PROXY"])
+
+    # disables tg->client traffic reencryption, faster but less secure
+    conf_dict.setdefault("FAST_MODE", True)
+
+    # enables some working modes
+    modes = conf_dict.get("MODES", {})
+
+    if "MODES" not in conf_dict:
+        modes.setdefault("classic", True)
+        modes.setdefault("secure", True)
+        modes.setdefault("tls", True)
+    else:
+        modes.setdefault("classic", False)
+        modes.setdefault("secure", False)
+        modes.setdefault("tls", False)
+
+    legacy_warning = False
+    if "SECURE_ONLY" in conf_dict:
+        legacy_warning = True
+        modes["classic"] = not bool(conf_dict["SECURE_ONLY"])
+
+    if "TLS_ONLY" in conf_dict:
+        legacy_warning = True
+        if conf_dict["TLS_ONLY"]:
+            modes["classic"] = False
+            modes["secure"] = False
+
+    if not modes["classic"] and not modes["secure"] and not modes["tls"]:
+        print_err("No known modes enabled, enabling tls-only mode")
+        modes["tls"] = True
+
+    if legacy_warning:
+        print_err("Legacy options SECURE_ONLY or TLS_ONLY detected")
+        print_err("Please use MODES in your config instead:")
+        print_err("MODES = {")
+        print_err('    "classic": %s,' % modes["classic"])
+        print_err('    "secure": %s,' % modes["secure"])
+        print_err('    "tls": %s' % modes["tls"])
+        print_err("}")
+
+    conf_dict["MODES"] = modes
+
+    # accept incoming connections only with proxy protocol v1/v2, useful for nginx and haproxy
+    conf_dict.setdefault("PROXY_PROTOCOL", False)
+
+    # set the tls domain for the proxy, has an influence only on starting message
+    conf_dict.setdefault("TLS_DOMAIN", "www.google.com")
+
+    # enable proxying bad clients to some host
+    conf_dict.setdefault("MASK", True)
+
+    # the next host to forward bad clients
+    conf_dict.setdefault("MASK_HOST", conf_dict["TLS_DOMAIN"])
+
+    # set the home domain for the proxy, has an influence only on the log message
+    conf_dict.setdefault("MY_DOMAIN", False)
+
+    # the next host's port to forward bad clients
+    conf_dict.setdefault("MASK_PORT", 443)
+
+    # use upstream SOCKS5 proxy
+    conf_dict.setdefault("SOCKS5_HOST", None)
+    conf_dict.setdefault("SOCKS5_PORT", None)
+    conf_dict.setdefault("SOCKS5_USER", None)
+    conf_dict.setdefault("SOCKS5_PASS", None)
+
+    if conf_dict["SOCKS5_HOST"] and conf_dict["SOCKS5_PORT"]:
+        # Disable the middle proxy if using socks, they are not compatible
+        conf_dict["USE_MIDDLE_PROXY"] = False
+
+    # user tcp connection limits, the mapping from name to the integer limit
+    # one client can create many tcp connections, up to 8
+    conf_dict.setdefault("USER_MAX_TCP_CONNS", {})
+
+    # expiration date for users in format of day/month/year
+    conf_dict.setdefault("USER_EXPIRATIONS", {})
+    for user in conf_dict["USER_EXPIRATIONS"]:
+        expiration = datetime.datetime.strptime(conf_dict["USER_EXPIRATIONS"][user], "%d/%m/%Y")
+        conf_dict["USER_EXPIRATIONS"][user] = expiration
+
+    # the data quota for user
+    conf_dict.setdefault("USER_DATA_QUOTA", {})
+
+    # length of used handshake randoms for active fingerprinting protection, zero to disable
+    conf_dict.setdefault("REPLAY_CHECK_LEN", 65536)
+
+    # accept clients with bad clocks. This reduces the protection against replay attacks
+    conf_dict.setdefault("IGNORE_TIME_SKEW", False)
+
+    # length of last client ip addresses for logging
+    conf_dict.setdefault("CLIENT_IPS_LEN", 131072)
+
+    # delay in seconds between stats printing
+    conf_dict.setdefault("STATS_PRINT_PERIOD", 600)
+
+    # delay in seconds between middle proxy info updates
+    conf_dict.setdefault("PROXY_INFO_UPDATE_PERIOD", 24*60*60)
+
+    # delay in seconds between time getting, zero means disabled
+    conf_dict.setdefault("GET_TIME_PERIOD", 10*60)
+
+    # delay in seconds between getting the length of certificate on the mask host
+    conf_dict.setdefault("GET_CERT_LEN_PERIOD", random.randrange(4*60*60, 6*60*60))
+
+    # max socket buffer size to the client direction, the more the faster, but more RAM hungry
+    # can be the tuple (low, users_margin, high) for the adaptive case. If no much users, use high
+    conf_dict.setdefault("TO_CLT_BUFSIZE", (16384, 100, 131072))
+
+    # max socket buffer size to the telegram servers direction, also can be the tuple
+    conf_dict.setdefault("TO_TG_BUFSIZE", 65536)
+
+    # keepalive period for clients in secs
+    conf_dict.setdefault("CLIENT_KEEPALIVE", 10*60)
+
+    # drop client after this timeout if the handshake fail
+    conf_dict.setdefault("CLIENT_HANDSHAKE_TIMEOUT", random.randrange(5, 15))
+
+    # if client doesn't confirm data for this number of seconds, it is dropped
+    conf_dict.setdefault("CLIENT_ACK_TIMEOUT", 5*60)
+
+    # telegram servers connect timeout in seconds
+    conf_dict.setdefault("TG_CONNECT_TIMEOUT", 10)
+
+    # drop connection if no data from telegram server for this many seconds
+    conf_dict.setdefault("TG_READ_TIMEOUT", 60)
+
+    # listen address for IPv4
+    conf_dict.setdefault("LISTEN_ADDR_IPV4", "0.0.0.0")
+
+    # listen address for IPv6
+    conf_dict.setdefault("LISTEN_ADDR_IPV6", "::")
+
+    # listen unix socket
+    conf_dict.setdefault("LISTEN_UNIX_SOCK", "")
+
+    # prometheus exporter listen port, use some random port here
+    conf_dict.setdefault("METRICS_PORT", None)
+
+    # prometheus listen addr ipv4
+    conf_dict.setdefault("METRICS_LISTEN_ADDR_IPV4", "0.0.0.0")
+
+    # prometheus listen addr ipv6
+    conf_dict.setdefault("METRICS_LISTEN_ADDR_IPV6", None)
+
+    # prometheus scrapers whitelist
+    conf_dict.setdefault("METRICS_WHITELIST", ["127.0.0.1", "::1"])
+
+    # export proxy link to prometheus
+    conf_dict.setdefault("METRICS_EXPORT_LINKS", False)
+
+    # default prefix for metrics
+    conf_dict.setdefault("METRICS_PREFIX", "mtprotoproxy_")
+
+    # allow access to config by attributes
+    config = type("config", (dict,), conf_dict)(conf_dict)
+
+
+def apply_upstream_proxy_settings():
+    # apply socks settings in place
+    if config.SOCKS5_HOST and config.SOCKS5_PORT:
+        import socks
+        print_err("Socket-proxy mode activated, it is incompatible with advertising and uvloop")
+        socks.set_default_proxy(socks.PROXY_TYPE_SOCKS5, config.SOCKS5_HOST, config.SOCKS5_PORT,
+                                username=config.SOCKS5_USER, password=config.SOCKS5_PASS)
+        if not hasattr(socket, "origsocket"):
+            socket.origsocket = socket.socket
+            socket.socket = socks.socksocket
+    elif hasattr(socket, "origsocket"):
+        socket.socket = socket.origsocket
+        del socket.origsocket
+
+
+def try_use_cryptography_module():
+    from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+    from cryptography.hazmat.backends import default_backend
+
+    class CryptographyEncryptorAdapter:
+        __slots__ = ('encryptor', 'decryptor')
+
+        def __init__(self, cipher):
+            self.encryptor = cipher.encryptor()
+            self.decryptor = cipher.decryptor()
+
+        def encrypt(self, data):
+            return self.encryptor.update(data)
+
+        def decrypt(self, data):
+            return self.decryptor.update(data)
+
+    def create_aes_ctr(key, iv):
+        iv_bytes = int.to_bytes(iv, 16, "big")
+        cipher = Cipher(algorithms.AES(key), modes.CTR(iv_bytes), default_backend())
+        return CryptographyEncryptorAdapter(cipher)
+
+    def create_aes_cbc(key, iv):
+        cipher = Cipher(algorithms.AES(key), modes.CBC(iv), default_backend())
+        return CryptographyEncryptorAdapter(cipher)
+
+    return create_aes_ctr, create_aes_cbc
+
+
+def try_use_pycrypto_or_pycryptodome_module():
+    from Crypto.Cipher import AES
+    from Crypto.Util import Counter
+
+    def create_aes_ctr(key, iv):
+        ctr = Counter.new(128, initial_value=iv)
+        return AES.new(key, AES.MODE_CTR, counter=ctr)
+
+    def create_aes_cbc(key, iv):
+        return AES.new(key, AES.MODE_CBC, iv)
+
+    return create_aes_ctr, create_aes_cbc
+
+
+def use_slow_bundled_cryptography_module():
+    import pyaes
+
+    msg = "To make the program a *lot* faster, please install cryptography module: "
+    msg += "pip install cryptography\n"
+    print(msg, flush=True, file=sys.stderr)
+
+    class BundledEncryptorAdapter:
+        __slots__ = ('mode', )
+
+        def __init__(self, mode):
+            self.mode = mode
+
+        def encrypt(self, data):
+            encrypter = pyaes.Encrypter(self.mode, pyaes.PADDING_NONE)
+            return encrypter.feed(data) + encrypter.feed()
+
+        def decrypt(self, data):
+            decrypter = pyaes.Decrypter(self.mode, pyaes.PADDING_NONE)
+            return decrypter.feed(data) + decrypter.feed()
+
+    def create_aes_ctr(key, iv):
+        ctr = pyaes.Counter(iv)
+        return pyaes.AESModeOfOperationCTR(key, ctr)
+
+    def create_aes_cbc(key, iv):
+        mode = pyaes.AESModeOfOperationCBC(key, iv)
+        return BundledEncryptorAdapter(mode)
+    return create_aes_ctr, create_aes_cbc
+
+
+try:
+    create_aes_ctr, create_aes_cbc = try_use_cryptography_module()
+except ImportError:
+    try:
+        create_aes_ctr, create_aes_cbc = try_use_pycrypto_or_pycryptodome_module()
+    except ImportError:
+        create_aes_ctr, create_aes_cbc = use_slow_bundled_cryptography_module()
+
+
+def print_err(*params):
+    print(*params, file=sys.stderr, flush=True)
+
+
+def ensure_users_in_user_stats():
+    global user_stats
+
+    for user in config.USERS:
+        user_stats[user].update()
+
+
+def init_proxy_start_time():
+    global proxy_start_time
+    proxy_start_time = time.time()
+
+
+def update_stats(**kw_stats):
+    global stats
+    stats.update(**kw_stats)
+
+
+def update_user_stats(user, **kw_stats):
+    global user_stats
+    user_stats[user].update(**kw_stats)
+
+
+def update_durations(duration):
+    global stats
+
+    for bucket in STAT_DURATION_BUCKETS:
+        if duration <= bucket:
+            break
+
+    update_stats(**{"connects_with_duration_le_%s" % str(bucket): 1})
+
+
+def get_curr_connects_count():
+    global user_stats
+
+    all_connects = 0
+    for user, stat in user_stats.items():
+        all_connects += stat["curr_connects"]
+    return all_connects
+
+
+def get_to_tg_bufsize():
+    if isinstance(config.TO_TG_BUFSIZE, int):
+        return config.TO_TG_BUFSIZE
+
+    low, margin, high = config.TO_TG_BUFSIZE
+    return high if get_curr_connects_count() < margin else low
+
+
+def get_to_clt_bufsize():
+    if isinstance(config.TO_CLT_BUFSIZE, int):
+        return config.TO_CLT_BUFSIZE
+
+    low, margin, high = config.TO_CLT_BUFSIZE
+    return high if get_curr_connects_count() < margin else low
+
+
+class MyRandom(random.Random):
+    def __init__(self):
+        super().__init__()
+        key = bytes([random.randrange(256) for i in range(32)])
+        iv = random.randrange(256**16)
+
+        self.encryptor = create_aes_ctr(key, iv)
+        self.buffer = bytearray()
+
+    def getrandbits(self, k):
+        numbytes = (k + 7) // 8
+        return int.from_bytes(self.getrandbytes(numbytes), 'big') >> (numbytes * 8 - k)
+
+    def getrandbytes(self, n):
+        CHUNK_SIZE = 512
+
+        while n > len(self.buffer):
+            data = int.to_bytes(super().getrandbits(CHUNK_SIZE*8), CHUNK_SIZE, "big")
+            self.buffer += self.encryptor.encrypt(data)
+
+        result = self.buffer[:n]
+        self.buffer = self.buffer[n:]
+        return bytes(result)
+
+
+myrandom = MyRandom()
+
+
+class TgConnectionPool:
+    MAX_CONNS_IN_POOL = 16
+
+    def __init__(self):
+        self.pools = {}
+
+    async def open_tg_connection(self, host, port, init_func=None):
+        task = asyncio.open_connection(host, port, limit=get_to_clt_bufsize())
+        reader_tgt, writer_tgt = await asyncio.wait_for(task, timeout=config.TG_CONNECT_TIMEOUT)
+
+        set_keepalive(writer_tgt.get_extra_info("socket"))
+        set_bufsizes(writer_tgt.get_extra_info("socket"), get_to_clt_bufsize(), get_to_tg_bufsize())
+
+        if init_func:
+            return await asyncio.wait_for(init_func(host, port, reader_tgt, writer_tgt),
+                                          timeout=config.TG_CONNECT_TIMEOUT)
+        return reader_tgt, writer_tgt
+
+    def is_conn_dead(self, reader, writer):
+        if writer.transport.is_closing():
+            return True
+        raw_reader = reader
+        while hasattr(raw_reader, 'upstream'):
+            raw_reader = raw_reader.upstream
+        if raw_reader.at_eof():
+            return True
+        return False
+
+    def register_host_port(self, host, port, init_func):
+        if (host, port, init_func) not in self.pools:
+            self.pools[(host, port, init_func)] = []
+
+        while len(self.pools[(host, port, init_func)]) < TgConnectionPool.MAX_CONNS_IN_POOL:
+            connect_task = asyncio.ensure_future(self.open_tg_connection(host, port, init_func))
+            self.pools[(host, port, init_func)].append(connect_task)
+
+    async def get_connection(self, host, port, init_func=None):
+        self.register_host_port(host, port, init_func)
+
+        ret = None
+        for task in self.pools[(host, port, init_func)][:]:
+            if task.done():
+                if task.exception():
+                    self.pools[(host, port, init_func)].remove(task)
+                    continue
+
+                reader, writer, *other = task.result()
+                if self.is_conn_dead(reader, writer):
+                    self.pools[(host, port, init_func)].remove(task)
+                    writer.transport.abort()
+                    continue
+
+                if not ret:
+                    self.pools[(host, port, init_func)].remove(task)
+                    ret = (reader, writer, *other)
+
+        self.register_host_port(host, port, init_func)
+        if ret:
+            return ret
+        return await self.open_tg_connection(host, port, init_func)
+
+
+tg_connection_pool = TgConnectionPool()
+
+
+class LayeredStreamReaderBase:
+    __slots__ = ("upstream", )
+
+    def __init__(self, upstream):
+        self.upstream = upstream
+
+    async def read(self, n):
+        return await self.upstream.read(n)
+
+    async def readexactly(self, n):
+        return await self.upstream.readexactly(n)
+
+
+class LayeredStreamWriterBase:
+    __slots__ = ("upstream", )
+
+    def __init__(self, upstream):
+        self.upstream = upstream
+
+    def write(self, data, extra={}):
+        return self.upstream.write(data)
+
+    def write_eof(self):
+        return self.upstream.write_eof()
+
+    async def drain(self):
+        return await self.upstream.drain()
+
+    def close(self):
+        return self.upstream.close()
+
+    def abort(self):
+        return self.upstream.transport.abort()
+
+    def get_extra_info(self, name):
+        return self.upstream.get_extra_info(name)
+
+    @property
+    def transport(self):
+        return self.upstream.transport
+
+
+class FakeTLSStreamReader(LayeredStreamReaderBase):
+    __slots__ = ('buf', )
+
+    def __init__(self, upstream):
+        self.upstream = upstream
+        self.buf = bytearray()
+
+    async def read(self, n, ignore_buf=False):
+        if self.buf and not ignore_buf:
+            data = self.buf
+            self.buf = bytearray()
+            return bytes(data)
+
+        while True:
+            tls_rec_type = await self.upstream.readexactly(1)
+            if not tls_rec_type:
+                return b""
+
+            if tls_rec_type not in [b"\x14", b"\x17"]:
+                print_err("BUG: bad tls type %s in FakeTLSStreamReader" % tls_rec_type)
+                return b""
+
+            version = await self.upstream.readexactly(2)
+            if version != b"\x03\x03":
+                print_err("BUG: unknown version %s in FakeTLSStreamReader" % version)
+                return b""
+
+            data_len = int.from_bytes(await self.upstream.readexactly(2), "big")
+            data = await self.upstream.readexactly(data_len)
+            if tls_rec_type == b"\x14":
+                continue
+            return data
+
+    async def readexactly(self, n):
+        while len(self.buf) < n:
+            tls_data = await self.read(1, ignore_buf=True)
+            if not tls_data:
+                return b""
+            self.buf += tls_data
+        data, self.buf = self.buf[:n], self.buf[n:]
+        return bytes(data)
+
+
+class FakeTLSStreamWriter(LayeredStreamWriterBase):
+    __slots__ = ()
+
+    def __init__(self, upstream):
+        self.upstream = upstream
+
+    def write(self, data, extra={}):
+        MAX_CHUNK_SIZE = 16384 + 24
+        for start in range(0, len(data), MAX_CHUNK_SIZE):
+            end = min(start+MAX_CHUNK_SIZE, len(data))
+            self.upstream.write(b"\x17\x03\x03" + int.to_bytes(end-start, 2, "big"))
+            self.upstream.write(data[start: end])
+        return len(data)
+
+
+class CryptoWrappedStreamReader(LayeredStreamReaderBase):
+    __slots__ = ('decryptor', 'block_size', 'buf')
+
+    def __init__(self, upstream, decryptor, block_size=1):
+        self.upstream = upstream
+        self.decryptor = decryptor
+        self.block_size = block_size
+        self.buf = bytearray()
+
+    async def read(self, n):
+        if self.buf:
+            ret = bytes(self.buf)
+            self.buf.clear()
+            return ret
+        else:
+            data = await self.upstream.read(n)
+            if not data:
+                return b""
+
+            needed_till_full_block = -len(data) % self.block_size
+            if needed_till_full_block > 0:
+                data += await self.upstream.readexactly(needed_till_full_block)
+            return self.decryptor.decrypt(data)
+
+    async def readexactly(self, n):
+        if n > len(self.buf):
+            to_read = n - len(self.buf)
+            needed_till_full_block = -to_read % self.block_size
+
+            to_read_block_aligned = to_read + needed_till_full_block
+            data = await self.upstream.readexactly(to_read_block_aligned)
+            self.buf += self.decryptor.decrypt(data)
+
+        ret = bytes(self.buf[:n])
+        self.buf = self.buf[n:]
+        return ret
+
+
+class CryptoWrappedStreamWriter(LayeredStreamWriterBase):
+    __slots__ = ('encryptor', 'block_size')
+
+    def __init__(self, upstream, encryptor, block_size=1):
+        self.upstream = upstream
+        self.encryptor = encryptor
+        self.block_size = block_size
+
+    def write(self, data, extra={}):
+        if len(data) % self.block_size != 0:
+            print_err("BUG: writing %d bytes not aligned to block size %d" % (
+                      len(data), self.block_size))
+            return 0
+        q = self.encryptor.encrypt(data)
+        return self.upstream.write(q)
+
+
+class MTProtoFrameStreamReader(LayeredStreamReaderBase):
+    __slots__ = ('seq_no', )
+
+    def __init__(self, upstream, seq_no=0):
+        self.upstream = upstream
+        self.seq_no = seq_no
+
+    async def read(self, buf_size):
+        msg_len_bytes = await self.upstream.readexactly(4)
+        msg_len = int.from_bytes(msg_len_bytes, "little")
+        # skip paddings
+        while msg_len == 4:
+            msg_len_bytes = await self.upstream.readexactly(4)
+            msg_len = int.from_bytes(msg_len_bytes, "little")
+
+        len_is_bad = (msg_len % len(PADDING_FILLER) != 0)
+        if not MIN_MSG_LEN <= msg_len <= MAX_MSG_LEN or len_is_bad:
+            print_err("msg_len is bad, closing connection", msg_len)
+            return b""
+
+        msg_seq_bytes = await self.upstream.readexactly(4)
+        msg_seq = int.from_bytes(msg_seq_bytes, "little", signed=True)
+        if msg_seq != self.seq_no:
+            print_err("unexpected seq_no")
+            return b""
+
+        self.seq_no += 1
+
+        data = await self.upstream.readexactly(msg_len - 4 - 4 - 4)
+        checksum_bytes = await self.upstream.readexactly(4)
+        checksum = int.from_bytes(checksum_bytes, "little")
+
+        computed_checksum = binascii.crc32(msg_len_bytes + msg_seq_bytes + data)
+        if computed_checksum != checksum:
+            return b""
+        return data
+
+
+class MTProtoFrameStreamWriter(LayeredStreamWriterBase):
+    __slots__ = ('seq_no', )
+
+    def __init__(self, upstream, seq_no=0):
+        self.upstream = upstream
+        self.seq_no = seq_no
+
+    def write(self, msg, extra={}):
+        len_bytes = int.to_bytes(len(msg) + 4 + 4 + 4, 4, "little")
+        seq_bytes = int.to_bytes(self.seq_no, 4, "little", signed=True)
+        self.seq_no += 1
+
+        msg_without_checksum = len_bytes + seq_bytes + msg
+        checksum = int.to_bytes(binascii.crc32(msg_without_checksum), 4, "little")
+
+        full_msg = msg_without_checksum + checksum
+        padding = PADDING_FILLER * ((-len(full_msg) % CBC_PADDING) // len(PADDING_FILLER))
+
+        return self.upstream.write(full_msg + padding)
+
+
+class MTProtoCompactFrameStreamReader(LayeredStreamReaderBase):
+    __slots__ = ()
+
+    async def read(self, buf_size):
+        msg_len_bytes = await self.upstream.readexactly(1)
+        msg_len = int.from_bytes(msg_len_bytes, "little")
+
+        extra = {"QUICKACK_FLAG": False}
+        if msg_len >= 0x80:
+            extra["QUICKACK_FLAG"] = True
+            msg_len -= 0x80
+
+        if msg_len == 0x7f:
+            msg_len_bytes = await self.upstream.readexactly(3)
+            msg_len = int.from_bytes(msg_len_bytes, "little")
+
+        msg_len *= 4
+
+        data = await self.upstream.readexactly(msg_len)
+
+        return data, extra
+
+
+class MTProtoCompactFrameStreamWriter(LayeredStreamWriterBase):
+    __slots__ = ()
+
+    def write(self, data, extra={}):
+        SMALL_PKT_BORDER = 0x7f
+        LARGE_PKT_BORDER = 256 ** 3
+
+        if len(data) % 4 != 0:
+            print_err("BUG: MTProtoFrameStreamWriter attempted to send msg with len", len(data))
+            return 0
+
+        if extra.get("SIMPLE_ACK"):
+            return self.upstream.write(data[::-1])
+
+        len_div_four = len(data) // 4
+
+        if len_div_four < SMALL_PKT_BORDER:
+            return self.upstream.write(bytes([len_div_four]) + data)
+        elif len_div_four < LARGE_PKT_BORDER:
+            return self.upstream.write(b'\x7f' + int.to_bytes(len_div_four, 3, 'little') + data)
+        else:
+            print_err("Attempted to send too large pkt len =", len(data))
+            return 0
+
+
+class MTProtoIntermediateFrameStreamReader(LayeredStreamReaderBase):
+    __slots__ = ()
+
+    async def read(self, buf_size):
+        msg_len_bytes = await self.upstream.readexactly(4)
+        msg_len = int.from_bytes(msg_len_bytes, "little")
+
+        extra = {}
+        if msg_len > 0x80000000:
+            extra["QUICKACK_FLAG"] = True
+            msg_len -= 0x80000000
+
+        data = await self.upstream.readexactly(msg_len)
+        return data, extra
+
+
+class MTProtoIntermediateFrameStreamWriter(LayeredStreamWriterBase):
+    __slots__ = ()
+
+    def write(self, data, extra={}):
+        if extra.get("SIMPLE_ACK"):
+            return self.upstream.write(data)
+        else:
+            return self.upstream.write(int.to_bytes(len(data), 4, 'little') + data)
+
+
+class MTProtoSecureIntermediateFrameStreamReader(LayeredStreamReaderBase):
+    __slots__ = ()
+
+    async def read(self, buf_size):
+        msg_len_bytes = await self.upstream.readexactly(4)
+        msg_len = int.from_bytes(msg_len_bytes, "little")
+
+        extra = {}
+        if msg_len > 0x80000000:
+            extra["QUICKACK_FLAG"] = True
+            msg_len -= 0x80000000
+
+        data = await self.upstream.readexactly(msg_len)
+
+        if msg_len % 4 != 0:
+            cut_border = msg_len - (msg_len % 4)
+            data = data[:cut_border]
+
+        return data, extra
+
+
+class MTProtoSecureIntermediateFrameStreamWriter(LayeredStreamWriterBase):
+    __slots__ = ()
+
+    def write(self, data, extra={}):
+        MAX_PADDING_LEN = 4
+        if extra.get("SIMPLE_ACK"):
+            # TODO: make this unpredictable
+            return self.upstream.write(data)
+        else:
+            padding_len = myrandom.randrange(MAX_PADDING_LEN)
+            padding = myrandom.getrandbytes(padding_len)
+            padded_data_len_bytes = int.to_bytes(len(data) + padding_len, 4, 'little')
+            return self.upstream.write(padded_data_len_bytes + data + padding)
+
+
+class ProxyReqStreamReader(LayeredStreamReaderBase):
+    __slots__ = ()
+
+    async def read(self, msg):
+        RPC_PROXY_ANS = b"\x0d\xda\x03\x44"
+        RPC_CLOSE_EXT = b"\xa2\x34\xb6\x5e"
+        RPC_SIMPLE_ACK = b"\x9b\x40\xac\x3b"
+        RPC_UNKNOWN = b'\xdf\xa2\x30\x57'
+
+        data = await self.upstream.read(1)
+
+        if len(data) < 4:
+            return b""
+
+        ans_type = data[:4]
+        if ans_type == RPC_CLOSE_EXT:
+            return b""
+
+        if ans_type == RPC_PROXY_ANS:
+            ans_flags, conn_id, conn_data = data[4:8], data[8:16], data[16:]
+            return conn_data
+
+        if ans_type == RPC_SIMPLE_ACK:
+            conn_id, confirm = data[4:12], data[12:16]
+            return confirm, {"SIMPLE_ACK": True}
+
+        if ans_type == RPC_UNKNOWN:
+            return b"", {"SKIP_SEND": True}
+
+        print_err("unknown rpc ans type:", ans_type)
+        return b"", {"SKIP_SEND": True}
+
+
+class ProxyReqStreamWriter(LayeredStreamWriterBase):
+    __slots__ = ('remote_ip_port', 'our_ip_port', 'out_conn_id', 'proto_tag')
+
+    def __init__(self, upstream, cl_ip, cl_port, my_ip, my_port, proto_tag):
+        self.upstream = upstream
+
+        if ":" not in cl_ip:
+            self.remote_ip_port = b"\x00" * 10 + b"\xff\xff"
+            self.remote_ip_port += socket.inet_pton(socket.AF_INET, cl_ip)
+        else:
+            self.remote_ip_port = socket.inet_pton(socket.AF_INET6, cl_ip)
+        self.remote_ip_port += int.to_bytes(cl_port, 4, "little")
+
+        if ":" not in my_ip:
+            self.our_ip_port = b"\x00" * 10 + b"\xff\xff"
+            self.our_ip_port += socket.inet_pton(socket.AF_INET, my_ip)
+        else:
+            self.our_ip_port = socket.inet_pton(socket.AF_INET6, my_ip)
+        self.our_ip_port += int.to_bytes(my_port, 4, "little")
+        self.out_conn_id = myrandom.getrandbytes(8)
+
+        self.proto_tag = proto_tag
+
+    def write(self, msg, extra={}):
+        RPC_PROXY_REQ = b"\xee\xf1\xce\x36"
+        EXTRA_SIZE = b"\x18\x00\x00\x00"
+        PROXY_TAG = b"\xae\x26\x1e\xdb"
+        FOUR_BYTES_ALIGNER = b"\x00\x00\x00"
+
+        FLAG_NOT_ENCRYPTED = 0x2
+        FLAG_HAS_AD_TAG = 0x8
+        FLAG_MAGIC = 0x1000
+        FLAG_EXTMODE2 = 0x20000
+        FLAG_PAD = 0x8000000
+        FLAG_INTERMEDIATE = 0x20000000
+        FLAG_ABRIDGED = 0x40000000
+        FLAG_QUICKACK = 0x80000000
+
+        if len(msg) % 4 != 0:
+            print_err("BUG: attempted to send msg with len %d" % len(msg))
+            return 0
+
+        flags = FLAG_MAGIC | FLAG_EXTMODE2
+
+        if self.proto_tag == PROTO_TAG_ABRIDGED:
+            flags |= FLAG_ABRIDGED
+        elif self.proto_tag == PROTO_TAG_INTERMEDIATE:
+            flags |= FLAG_INTERMEDIATE
+        elif self.proto_tag == PROTO_TAG_SECURE:
+            flags |= FLAG_INTERMEDIATE | FLAG_PAD
+
+        if extra.get("QUICKACK_FLAG"):
+            flags |= FLAG_QUICKACK
+
+        if msg.startswith(b"\x00" * 8):
+            flags |= FLAG_NOT_ENCRYPTED
+
+        if len(config.AD_TAG) == 16:
+            flags |= FLAG_HAS_AD_TAG
+
+        full_msg = bytearray()
+        full_msg += RPC_PROXY_REQ + int.to_bytes(flags, 4, "little") + self.out_conn_id
+        full_msg += self.remote_ip_port + self.our_ip_port
+        if len(config.AD_TAG) == 16:
+            full_msg += EXTRA_SIZE + PROXY_TAG
+            full_msg += bytes([len(config.AD_TAG)]) + config.AD_TAG + FOUR_BYTES_ALIGNER
+        full_msg += msg
+
+        return self.upstream.write(full_msg)
+
+
+def try_setsockopt(sock, level, option, value):
+    try:
+        sock.setsockopt(level, option, value)
+    except OSError as E:
+        pass
+
+
+def set_keepalive(sock, interval=40, attempts=5):
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+    if hasattr(socket, "TCP_KEEPIDLE"):
+        try_setsockopt(sock, socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, interval)
+    if hasattr(socket, "TCP_KEEPINTVL"):
+        try_setsockopt(sock, socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, interval)
+    if hasattr(socket, "TCP_KEEPCNT"):
+        try_setsockopt(sock, socket.IPPROTO_TCP, socket.TCP_KEEPCNT, attempts)
+
+
+def set_ack_timeout(sock, timeout):
+    if hasattr(socket, "TCP_USER_TIMEOUT"):
+        try_setsockopt(sock, socket.IPPROTO_TCP, socket.TCP_USER_TIMEOUT, timeout*1000)
+
+
+def set_bufsizes(sock, recv_buf, send_buf):
+    try_setsockopt(sock, socket.SOL_SOCKET, socket.SO_RCVBUF, recv_buf)
+    try_setsockopt(sock, socket.SOL_SOCKET, socket.SO_SNDBUF, send_buf)
+
+
+def set_instant_rst(sock):
+    INSTANT_RST = b"\x01\x00\x00\x00\x00\x00\x00\x00"
+    if hasattr(socket, "SO_LINGER"):
+        try_setsockopt(sock, socket.SOL_SOCKET, socket.SO_LINGER, INSTANT_RST)
+
+
+def gen_x25519_public_key():
+    # generates some number which has square root by modulo P
+    P = 2**255 - 19
+    n = myrandom.randrange(P)
+    return int.to_bytes((n*n) % P, length=32, byteorder="little")
+
+
+async def connect_reader_to_writer(reader, writer):
+    BUF_SIZE = 8192
+    try:
+        while True:
+            data = await reader.read(BUF_SIZE)
+
+            if not data:
+                if not writer.transport.is_closing():
+                    writer.write_eof()
+                    await writer.drain()
+                return
+
+            writer.write(data)
+            await writer.drain()
+    except (OSError, asyncio.IncompleteReadError) as e:
+        pass
+
+
+async def handle_bad_client(reader_clt, writer_clt, handshake):
+    BUF_SIZE = 8192
+    CONNECT_TIMEOUT = 5
+
+    global mask_host_cached_ip
+
+    update_stats(connects_bad=1)
+
+    if writer_clt.transport.is_closing():
+        return
+
+    set_bufsizes(writer_clt.get_extra_info("socket"), BUF_SIZE, BUF_SIZE)
+
+    if not config.MASK or handshake is None:
+        while await reader_clt.read(BUF_SIZE):
+            # just consume all the data
+            pass
+        return
+
+    writer_srv = None
+    try:
+        host = mask_host_cached_ip or config.MASK_HOST
+        task = asyncio.open_connection(host, config.MASK_PORT, limit=BUF_SIZE)
+        reader_srv, writer_srv = await asyncio.wait_for(task, timeout=CONNECT_TIMEOUT)
+        if not mask_host_cached_ip:
+            mask_host_cached_ip = writer_srv.get_extra_info("peername")[0]
+        writer_srv.write(handshake)
+        await writer_srv.drain()
+
+        srv_to_clt = connect_reader_to_writer(reader_srv, writer_clt)
+        clt_to_srv = connect_reader_to_writer(reader_clt, writer_srv)
+        task_srv_to_clt = asyncio.ensure_future(srv_to_clt)
+        task_clt_to_srv = asyncio.ensure_future(clt_to_srv)
+
+        await asyncio.wait([task_srv_to_clt, task_clt_to_srv], return_when=asyncio.FIRST_COMPLETED)
+
+        task_srv_to_clt.cancel()
+        task_clt_to_srv.cancel()
+
+        if writer_clt.transport.is_closing():
+            return
+
+        # if the server closed the connection with RST or FIN-RST, copy them to the client
+        if not writer_srv.transport.is_closing():
+            # workaround for uvloop, it doesn't fire exceptions on write_eof
+            sock = writer_srv.get_extra_info('socket')
+            raw_sock = socket.socket(sock.family, sock.type, sock.proto, sock.fileno())
+            try:
+                raw_sock.shutdown(socket.SHUT_WR)
+            except OSError as E:
+                set_instant_rst(writer_clt.get_extra_info("socket"))
+            finally:
+                raw_sock.detach()
+        else:
+            set_instant_rst(writer_clt.get_extra_info("socket"))
+    except ConnectionRefusedError as E:
+        return
+    except (OSError, asyncio.TimeoutError) as E:
+        return
+    finally:
+        if writer_srv is not None:
+            writer_srv.transport.abort()
+
+
+async def handle_fake_tls_handshake(handshake, reader, writer, peer):
+    global used_handshakes
+    global client_ips
+    global last_client_ips
+    global last_clients_with_time_skew
+    global last_clients_with_same_handshake
+    global fake_cert_len
+
+    TIME_SKEW_MIN = -20 * 60
+    TIME_SKEW_MAX = 10 * 60
+
+    TLS_VERS = b"\x03\x03"
+    TLS_CIPHERSUITE = b"\x13\x01"
+    TLS_CHANGE_CIPHER = b"\x14" + TLS_VERS + b"\x00\x01\x01"
+    TLS_APP_HTTP2_HDR = b"\x17" + TLS_VERS
+
+    DIGEST_LEN = 32
+    DIGEST_HALFLEN = 16
+    DIGEST_POS = 11
+
+    SESSION_ID_LEN_POS = DIGEST_POS + DIGEST_LEN
+    SESSION_ID_POS = SESSION_ID_LEN_POS + 1
+
+    tls_extensions = b"\x00\x2e" + b"\x00\x33\x00\x24" + b"\x00\x1d\x00\x20"
+    tls_extensions += gen_x25519_public_key() + b"\x00\x2b\x00\x02\x03\x04"
+
+    digest = handshake[DIGEST_POS:DIGEST_POS+DIGEST_LEN]
+
+    if digest[:DIGEST_HALFLEN] in used_handshakes:
+        last_clients_with_same_handshake[peer[0]] += 1
+        return False
+
+    sess_id_len = handshake[SESSION_ID_LEN_POS]
+    sess_id = handshake[SESSION_ID_POS:SESSION_ID_POS+sess_id_len]
+
+    for user in config.USERS:
+        secret = bytes.fromhex(config.USERS[user])
+
+        msg = handshake[:DIGEST_POS] + b"\x00"*DIGEST_LEN + handshake[DIGEST_POS+DIGEST_LEN:]
+        computed_digest = hmac.new(secret, msg, digestmod=hashlib.sha256).digest()
+
+        xored_digest = bytes(digest[i] ^ computed_digest[i] for i in range(DIGEST_LEN))
+        digest_good = xored_digest.startswith(b"\x00" * (DIGEST_LEN-4))
+
+        if not digest_good:
+            continue
+
+        timestamp = int.from_bytes(xored_digest[-4:], "little")
+        client_time_is_ok = TIME_SKEW_MIN < time.time() - timestamp < TIME_SKEW_MAX
+
+        # some clients fail to read unix time and send the time since boot instead
+        client_time_is_small = timestamp < 60*60*24*1000
+        accept_bad_time = config.IGNORE_TIME_SKEW or is_time_skewed or client_time_is_small
+
+        if not client_time_is_ok and not accept_bad_time:
+            last_clients_with_time_skew[peer[0]] = (time.time() - timestamp) // 60
+            continue
+
+        http_data = myrandom.getrandbytes(fake_cert_len)
+
+        srv_hello = TLS_VERS + b"\x00"*DIGEST_LEN + bytes([sess_id_len]) + sess_id
+        srv_hello += TLS_CIPHERSUITE + b"\x00" + tls_extensions
+
+        hello_pkt = b"\x16" + TLS_VERS + int.to_bytes(len(srv_hello) + 4, 2, "big")
+        hello_pkt += b"\x02" + int.to_bytes(len(srv_hello), 3, "big") + srv_hello
+        hello_pkt += TLS_CHANGE_CIPHER + TLS_APP_HTTP2_HDR
+        hello_pkt += int.to_bytes(len(http_data), 2, "big") + http_data
+
+        computed_digest = hmac.new(secret, msg=digest+hello_pkt, digestmod=hashlib.sha256).digest()
+        hello_pkt = hello_pkt[:DIGEST_POS] + computed_digest + hello_pkt[DIGEST_POS+DIGEST_LEN:]
+
+        writer.write(hello_pkt)
+        await writer.drain()
+
+        if config.REPLAY_CHECK_LEN > 0:
+            while len(used_handshakes) >= config.REPLAY_CHECK_LEN:
+                used_handshakes.popitem(last=False)
+            used_handshakes[digest[:DIGEST_HALFLEN]] = True
+
+        if config.CLIENT_IPS_LEN > 0:
+            while len(client_ips) >= config.CLIENT_IPS_LEN:
+                client_ips.popitem(last=False)
+            if peer[0] not in client_ips:
+                client_ips[peer[0]] = True
+                last_client_ips[peer[0]] = True
+
+        reader = FakeTLSStreamReader(reader)
+        writer = FakeTLSStreamWriter(writer)
+        return reader, writer
+
+    return False
+
+
+async def handle_proxy_protocol(reader, peer=None):
+    PROXY_SIGNATURE = b"PROXY "
+    PROXY_MIN_LEN = 6
+    PROXY_TCP4 = b"TCP4"
+    PROXY_TCP6 = b"TCP6"
+    PROXY_UNKNOWN = b"UNKNOWN"
+
+    PROXY2_SIGNATURE = b"\x0d\x0a\x0d\x0a\x00\x0d\x0a\x51\x55\x49\x54\x0a"
+    PROXY2_MIN_LEN = 16
+    PROXY2_AF_UNSPEC = 0x0
+    PROXY2_AF_INET = 0x1
+    PROXY2_AF_INET6 = 0x2
+
+    header = await reader.readexactly(PROXY_MIN_LEN)
+    if header.startswith(PROXY_SIGNATURE):
+        # proxy header v1
+        header += await reader.readuntil(b"\r\n")
+        _, proxy_fam, *proxy_addr = header[:-2].split(b" ")
+        if proxy_fam in (PROXY_TCP4, PROXY_TCP6):
+            if len(proxy_addr) == 4:
+                src_addr = proxy_addr[0].decode('ascii')
+                src_port = int(proxy_addr[2].decode('ascii'))
+                return (src_addr, src_port)
+        elif proxy_fam == PROXY_UNKNOWN:
+            return peer
+        return False
+
+    header += await reader.readexactly(PROXY2_MIN_LEN - PROXY_MIN_LEN)
+    if header.startswith(PROXY2_SIGNATURE):
+        # proxy header v2
+        proxy_ver = header[12]
+        if proxy_ver & 0xf0 != 0x20:
+            return False
+        proxy_len = int.from_bytes(header[14:16], "big")
+        proxy_addr = await reader.readexactly(proxy_len)
+        if proxy_ver == 0x21:
+            proxy_fam = header[13] >> 4
+            if proxy_fam == PROXY2_AF_INET:
+                if proxy_len >= (4 + 2)*2:
+                    src_addr = socket.inet_ntop(socket.AF_INET, proxy_addr[:4])
+                    src_port = int.from_bytes(proxy_addr[8:10], "big")
+                    return (src_addr, src_port)
+            elif proxy_fam == PROXY2_AF_INET6:
+                if proxy_len >= (16 + 2)*2:
+                    src_addr = socket.inet_ntop(socket.AF_INET6, proxy_addr[:16])
+                    src_port = int.from_bytes(proxy_addr[32:34], "big")
+                    return (src_addr, src_port)
+            elif proxy_fam == PROXY2_AF_UNSPEC:
+                return peer
+        elif proxy_ver == 0x20:
+            return peer
+
+    return False
+
+
+async def handle_handshake(reader, writer):
+    global used_handshakes
+    global client_ips
+    global last_client_ips
+    global last_clients_with_same_handshake
+
+    TLS_START_BYTES = b"\x16\x03\x01"
+
+    if writer.transport.is_closing() or writer.get_extra_info("peername") is None:
+        return False
+
+    peer = writer.get_extra_info("peername")[:2]
+    if not peer:
+        peer = ("unknown ip", 0)
+
+    if config.PROXY_PROTOCOL:
+        ip = peer[0] if peer else "unknown ip"
+        peer = await handle_proxy_protocol(reader, peer)
+        if not peer:
+            print_err("Client from %s sent bad proxy protocol headers" % ip)
+            await handle_bad_client(reader, writer, None)
+            return False
+
+    is_tls_handshake = True
+    handshake = b""
+    for expected_byte in TLS_START_BYTES:
+        handshake += await reader.readexactly(1)
+        if handshake[-1] != expected_byte:
+            is_tls_handshake = False
+            break
+
+    if is_tls_handshake:
+        handshake += await reader.readexactly(2)
+        tls_handshake_len = int.from_bytes(handshake[-2:], "big")
+        if tls_handshake_len < 512:
+            is_tls_handshake = False
+
+    if is_tls_handshake:
+        handshake += await reader.readexactly(tls_handshake_len)
+        tls_handshake_result = await handle_fake_tls_handshake(handshake, reader, writer, peer)
+
+        if not tls_handshake_result:
+            await handle_bad_client(reader, writer, handshake)
+            return False
+        reader, writer = tls_handshake_result
+        handshake = await reader.readexactly(HANDSHAKE_LEN)
+    else:
+        if not config.MODES["classic"] and not config.MODES["secure"]:
+            await handle_bad_client(reader, writer, handshake)
+            return False
+        handshake += await reader.readexactly(HANDSHAKE_LEN - len(handshake))
+
+    dec_prekey_and_iv = handshake[SKIP_LEN:SKIP_LEN+PREKEY_LEN+IV_LEN]
+    dec_prekey, dec_iv = dec_prekey_and_iv[:PREKEY_LEN], dec_prekey_and_iv[PREKEY_LEN:]
+    enc_prekey_and_iv = handshake[SKIP_LEN:SKIP_LEN+PREKEY_LEN+IV_LEN][::-1]
+    enc_prekey, enc_iv = enc_prekey_and_iv[:PREKEY_LEN], enc_prekey_and_iv[PREKEY_LEN:]
+
+    if dec_prekey_and_iv in used_handshakes:
+        last_clients_with_same_handshake[peer[0]] += 1
+        await handle_bad_client(reader, writer, handshake)
+        return False
+
+    for user in config.USERS:
+        secret = bytes.fromhex(config.USERS[user])
+
+        dec_key = hashlib.sha256(dec_prekey + secret).digest()
+        decryptor = create_aes_ctr(key=dec_key, iv=int.from_bytes(dec_iv, "big"))
+
+        enc_key = hashlib.sha256(enc_prekey + secret).digest()
+        encryptor = create_aes_ctr(key=enc_key, iv=int.from_bytes(enc_iv, "big"))
+
+        decrypted = decryptor.decrypt(handshake)
+
+        proto_tag = decrypted[PROTO_TAG_POS:PROTO_TAG_POS+4]
+        if proto_tag not in (PROTO_TAG_ABRIDGED, PROTO_TAG_INTERMEDIATE, PROTO_TAG_SECURE):
+            continue
+
+        if proto_tag == PROTO_TAG_SECURE:
+            if is_tls_handshake and not config.MODES["tls"]:
+                continue
+            if not is_tls_handshake and not config.MODES["secure"]:
+                continue
+        else:
+            if not config.MODES["classic"]:
+                continue
+
+        dc_idx = int.from_bytes(decrypted[DC_IDX_POS:DC_IDX_POS+2], "little", signed=True)
+
+        if config.REPLAY_CHECK_LEN > 0:
+            while len(used_handshakes) >= config.REPLAY_CHECK_LEN:
+                used_handshakes.popitem(last=False)
+            used_handshakes[dec_prekey_and_iv] = True
+
+        if config.CLIENT_IPS_LEN > 0:
+            while len(client_ips) >= config.CLIENT_IPS_LEN:
+                client_ips.popitem(last=False)
+            if peer[0] not in client_ips:
+                client_ips[peer[0]] = True
+                last_client_ips[peer[0]] = True
+
+        reader = CryptoWrappedStreamReader(reader, decryptor)
+        writer = CryptoWrappedStreamWriter(writer, encryptor)
+        return reader, writer, proto_tag, user, dc_idx, enc_key + enc_iv, peer
+
+    await handle_bad_client(reader, writer, handshake)
+    return False
+
+
+async def do_direct_handshake(proto_tag, dc_idx, dec_key_and_iv=None):
+    RESERVED_NONCE_FIRST_CHARS = [b"\xef"]
+    RESERVED_NONCE_BEGININGS = [b"\x48\x45\x41\x44", b"\x50\x4F\x53\x54",
+                                b"\x47\x45\x54\x20", b"\xee\xee\xee\xee",
+                                b"\xdd\xdd\xdd\xdd", b"\x16\x03\x01\x02"]
+    RESERVED_NONCE_CONTINUES = [b"\x00\x00\x00\x00"]
+
+    global my_ip_info
+    global tg_connection_pool
+
+    dc_idx = abs(dc_idx) - 1
+
+    if my_ip_info["ipv6"] and (config.PREFER_IPV6 or not my_ip_info["ipv4"]):
+        if not 0 <= dc_idx < len(TG_DATACENTERS_V6):
+            return False
+        dc = TG_DATACENTERS_V6[dc_idx]
+    else:
+        if not 0 <= dc_idx < len(TG_DATACENTERS_V4):
+            return False
+        dc = TG_DATACENTERS_V4[dc_idx]
+
+    try:
+        reader_tgt, writer_tgt = await tg_connection_pool.get_connection(dc, TG_DATACENTER_PORT)
+    except ConnectionRefusedError as E:
+        print_err("Got connection refused while trying to connect to", dc, TG_DATACENTER_PORT)
+        return False
+    except ConnectionAbortedError as E:
+        print_err("The Telegram server connection is bad: %d (%s %s) %s" % (dc_idx, dc, TG_DATACENTER_PORT, E))
+        return False
+    except (OSError, asyncio.TimeoutError) as E:
+        print_err("Unable to connect to", dc, TG_DATACENTER_PORT)
+        return False
+
+    while True:
+        rnd = bytearray(myrandom.getrandbytes(HANDSHAKE_LEN))
+        if rnd[:1] in RESERVED_NONCE_FIRST_CHARS:
+            continue
+        if rnd[:4] in RESERVED_NONCE_BEGININGS:
+            continue
+        if rnd[4:8] in RESERVED_NONCE_CONTINUES:
+            continue
+        break
+
+    rnd[PROTO_TAG_POS:PROTO_TAG_POS+4] = proto_tag
+
+    if dec_key_and_iv:
+        rnd[SKIP_LEN:SKIP_LEN+KEY_LEN+IV_LEN] = dec_key_and_iv[::-1]
+
+    rnd = bytes(rnd)
+
+    dec_key_and_iv = rnd[SKIP_LEN:SKIP_LEN+KEY_LEN+IV_LEN][::-1]
+    dec_key, dec_iv = dec_key_and_iv[:KEY_LEN], dec_key_and_iv[KEY_LEN:]
+    decryptor = create_aes_ctr(key=dec_key, iv=int.from_bytes(dec_iv, "big"))
+
+    enc_key_and_iv = rnd[SKIP_LEN:SKIP_LEN+KEY_LEN+IV_LEN]
+    enc_key, enc_iv = enc_key_and_iv[:KEY_LEN], enc_key_and_iv[KEY_LEN:]
+    encryptor = create_aes_ctr(key=enc_key, iv=int.from_bytes(enc_iv, "big"))
+
+    rnd_enc = rnd[:PROTO_TAG_POS] + encryptor.encrypt(rnd)[PROTO_TAG_POS:]
+
+    writer_tgt.write(rnd_enc)
+    await writer_tgt.drain()
+
+    reader_tgt = CryptoWrappedStreamReader(reader_tgt, decryptor)
+    writer_tgt = CryptoWrappedStreamWriter(writer_tgt, encryptor)
+
+    return reader_tgt, writer_tgt
+
+
+def get_middleproxy_aes_key_and_iv(nonce_srv, nonce_clt, clt_ts, srv_ip, clt_port, purpose,
+                                   clt_ip, srv_port, middleproxy_secret, clt_ipv6=None,
+                                   srv_ipv6=None):
+    EMPTY_IP = b"\x00\x00\x00\x00"
+
+    if not clt_ip or not srv_ip:
+        clt_ip = EMPTY_IP
+        srv_ip = EMPTY_IP
+
+    s = bytearray()
+    s += nonce_srv + nonce_clt + clt_ts + srv_ip + clt_port + purpose + clt_ip + srv_port
+    s += middleproxy_secret + nonce_srv
+
+    if clt_ipv6 and srv_ipv6:
+        s += clt_ipv6 + srv_ipv6
+
+    s += nonce_clt
+
+    md5_sum = hashlib.md5(s[1:]).digest()
+    sha1_sum = hashlib.sha1(s).digest()
+
+    key = md5_sum[:12] + sha1_sum
+    iv = hashlib.md5(s[2:]).digest()
+    return key, iv
+
+
+async def middleproxy_handshake(host, port, reader_tgt, writer_tgt):
+    """ The most logic of middleproxy handshake, launched in pool """
+    START_SEQ_NO = -2
+    NONCE_LEN = 16
+
+    RPC_HANDSHAKE = b"\xf5\xee\x82\x76"
+    RPC_NONCE = b"\xaa\x87\xcb\x7a"
+    # pass as consts to simplify code
+    RPC_FLAGS = b"\x00\x00\x00\x00"
+    CRYPTO_AES = b"\x01\x00\x00\x00"
+
+    RPC_NONCE_ANS_LEN = 32
+    RPC_HANDSHAKE_ANS_LEN = 32
+
+    writer_tgt = MTProtoFrameStreamWriter(writer_tgt, START_SEQ_NO)
+    key_selector = PROXY_SECRET[:4]
+    crypto_ts = int.to_bytes(int(time.time()) % (256**4), 4, "little")
+
+    nonce = myrandom.getrandbytes(NONCE_LEN)
+
+    msg = RPC_NONCE + key_selector + CRYPTO_AES + crypto_ts + nonce
+
+    writer_tgt.write(msg)
+    await writer_tgt.drain()
+
+    reader_tgt = MTProtoFrameStreamReader(reader_tgt, START_SEQ_NO)
+    ans = await reader_tgt.read(get_to_clt_bufsize())
+
+    if len(ans) != RPC_NONCE_ANS_LEN:
+        raise ConnectionAbortedError("bad rpc answer length")
+
+    rpc_type, rpc_key_selector, rpc_schema, rpc_crypto_ts, rpc_nonce = (
+        ans[:4], ans[4:8], ans[8:12], ans[12:16], ans[16:32]
+    )
+
+    if rpc_type != RPC_NONCE or rpc_key_selector != key_selector or rpc_schema != CRYPTO_AES:
+        raise ConnectionAbortedError("bad rpc answer")
+
+    # get keys
+    tg_ip, tg_port = writer_tgt.upstream.get_extra_info('peername')[:2]
+    my_ip, my_port = writer_tgt.upstream.get_extra_info('sockname')[:2]
+
+    use_ipv6_tg = (":" in tg_ip)
+
+    if not use_ipv6_tg:
+        if my_ip_info["ipv4"]:
+            # prefer global ip settings to work behind NAT
+            my_ip = my_ip_info["ipv4"]
+
+        tg_ip_bytes = socket.inet_pton(socket.AF_INET, tg_ip)[::-1]
+        my_ip_bytes = socket.inet_pton(socket.AF_INET, my_ip)[::-1]
+
+        tg_ipv6_bytes = None
+        my_ipv6_bytes = None
+    else:
+        if my_ip_info["ipv6"]:
+            my_ip = my_ip_info["ipv6"]
+
+        tg_ip_bytes = None
+        my_ip_bytes = None
+
+        tg_ipv6_bytes = socket.inet_pton(socket.AF_INET6, tg_ip)
+        my_ipv6_bytes = socket.inet_pton(socket.AF_INET6, my_ip)
+
+    tg_port_bytes = int.to_bytes(tg_port, 2, "little")
+    my_port_bytes = int.to_bytes(my_port, 2, "little")
+
+    enc_key, enc_iv = get_middleproxy_aes_key_and_iv(
+        nonce_srv=rpc_nonce, nonce_clt=nonce, clt_ts=crypto_ts, srv_ip=tg_ip_bytes,
+        clt_port=my_port_bytes, purpose=b"CLIENT", clt_ip=my_ip_bytes, srv_port=tg_port_bytes,
+        middleproxy_secret=PROXY_SECRET, clt_ipv6=my_ipv6_bytes, srv_ipv6=tg_ipv6_bytes)
+
+    dec_key, dec_iv = get_middleproxy_aes_key_and_iv(
+        nonce_srv=rpc_nonce, nonce_clt=nonce, clt_ts=crypto_ts, srv_ip=tg_ip_bytes,
+        clt_port=my_port_bytes, purpose=b"SERVER", clt_ip=my_ip_bytes, srv_port=tg_port_bytes,
+        middleproxy_secret=PROXY_SECRET, clt_ipv6=my_ipv6_bytes, srv_ipv6=tg_ipv6_bytes)
+
+    encryptor = create_aes_cbc(key=enc_key, iv=enc_iv)
+    decryptor = create_aes_cbc(key=dec_key, iv=dec_iv)
+
+    SENDER_PID = b"IPIPPRPDTIME"
+    PEER_PID = b"IPIPPRPDTIME"
+
+    # TODO: pass client ip and port here for statistics
+    handshake = RPC_HANDSHAKE + RPC_FLAGS + SENDER_PID + PEER_PID
+
+    writer_tgt.upstream = CryptoWrappedStreamWriter(writer_tgt.upstream, encryptor, block_size=16)
+    writer_tgt.write(handshake)
+    await writer_tgt.drain()
+
+    reader_tgt.upstream = CryptoWrappedStreamReader(reader_tgt.upstream, decryptor, block_size=16)
+
+    handshake_ans = await reader_tgt.read(1)
+    if len(handshake_ans) != RPC_HANDSHAKE_ANS_LEN:
+        raise ConnectionAbortedError("bad rpc handshake answer length")
+
+    handshake_type, handshake_flags, handshake_sender_pid, handshake_peer_pid = (
+        handshake_ans[:4], handshake_ans[4:8], handshake_ans[8:20], handshake_ans[20:32])
+    if handshake_type != RPC_HANDSHAKE or handshake_peer_pid != SENDER_PID:
+        raise ConnectionAbortedError("bad rpc handshake answer")
+
+    return reader_tgt, writer_tgt, my_ip, my_port
+
+
+async def do_middleproxy_handshake(proto_tag, dc_idx, cl_ip, cl_port):
+    global my_ip_info
+    global tg_connection_pool
+
+    use_ipv6_tg = (my_ip_info["ipv6"] and (config.PREFER_IPV6 or not my_ip_info["ipv4"]))
+
+    if use_ipv6_tg:
+        if dc_idx not in TG_MIDDLE_PROXIES_V6:
+            return False
+        proxies = list(TG_MIDDLE_PROXIES_V6[dc_idx])
+    else:
+        if dc_idx not in TG_MIDDLE_PROXIES_V4:
+            return False
+        proxies = list(TG_MIDDLE_PROXIES_V4[dc_idx])
+
+    myrandom.shuffle(proxies)
+
+    for addr, port in proxies:
+        try:
+            reader_tgt, writer_tgt, my_ip, my_port = await tg_connection_pool.get_connection(
+                addr, port, middleproxy_handshake)
+            break
+        except ConnectionRefusedError as E:
+            print_err("The Telegram server %d (%s %s) is refusing connections, trying next" % (dc_idx, addr, port))
+        except ConnectionAbortedError as E:
+            print_err("The Telegram server connection is bad: %d (%s %s) %s, trying next" % (dc_idx, addr, port, E))
+        except (OSError, asyncio.TimeoutError) as E:
+            print_err("Unable to connect to the Telegram server %d (%s %s), trying next" % (dc_idx, addr, port))
+    else:
+        return False
+
+    writer_tgt = ProxyReqStreamWriter(writer_tgt, cl_ip, cl_port, my_ip, my_port, proto_tag)
+    reader_tgt = ProxyReqStreamReader(reader_tgt)
+
+    return reader_tgt, writer_tgt
+
+
+async def tg_connect_reader_to_writer(rd, wr, user, rd_buf_size, is_upstream):
+    try:
+        while True:
+            if not is_upstream:
+                data = await asyncio.wait_for(rd.read(rd_buf_size),
+                                              timeout=config.TG_READ_TIMEOUT)
+            else:
+                data = await rd.read(rd_buf_size)
+            if isinstance(data, tuple):
+                data, extra = data
+            else:
+                extra = {}
+
+            if extra.get("SKIP_SEND"):
+                continue
+
+            if not data:
+                wr.write_eof()
+                await wr.drain()
+                return
+            else:
+                if is_upstream:
+                    update_user_stats(user, octets_from_client=len(data), msgs_from_client=1)
+                else:
+                    update_user_stats(user, octets_to_client=len(data), msgs_to_client=1)
+
+                wr.write(data, extra)
+                await wr.drain()
+    except (OSError, asyncio.IncompleteReadError, asyncio.TimeoutError) as e:
+        # print_err(e)
+        pass
+
+
+async def handle_client(reader_clt, writer_clt):
+    set_keepalive(writer_clt.get_extra_info("socket"), config.CLIENT_KEEPALIVE, attempts=3)
+    set_ack_timeout(writer_clt.get_extra_info("socket"), config.CLIENT_ACK_TIMEOUT)
+    set_bufsizes(writer_clt.get_extra_info("socket"), get_to_tg_bufsize(), get_to_clt_bufsize())
+
+    update_stats(connects_all=1)
+
+    try:
+        clt_data = await asyncio.wait_for(handle_handshake(reader_clt, writer_clt),
+                                          timeout=config.CLIENT_HANDSHAKE_TIMEOUT)
+    except asyncio.TimeoutError:
+        update_stats(handshake_timeouts=1)
+        return
+
+    if not clt_data:
+        return
+
+    reader_clt, writer_clt, proto_tag, user, dc_idx, enc_key_and_iv, peer = clt_data
+    cl_ip, cl_port = peer
+
+    update_user_stats(user, connects=1)
+
+    connect_directly = (not config.USE_MIDDLE_PROXY or disable_middle_proxy)
+
+    if connect_directly:
+        if config.FAST_MODE:
+            tg_data = await do_direct_handshake(proto_tag, dc_idx, dec_key_and_iv=enc_key_and_iv)
+        else:
+            tg_data = await do_direct_handshake(proto_tag, dc_idx)
+    else:
+        tg_data = await do_middleproxy_handshake(proto_tag, dc_idx, cl_ip, cl_port)
+
+    if not tg_data:
+        return
+
+    reader_tg, writer_tg = tg_data
+
+    if connect_directly and config.FAST_MODE:
+        class FakeEncryptor:
+            def encrypt(self, data):
+                return data
+
+        class FakeDecryptor:
+            def decrypt(self, data):
+                return data
+
+        reader_tg.decryptor = FakeDecryptor()
+        writer_clt.encryptor = FakeEncryptor()
+
+    if not connect_directly:
+        if proto_tag == PROTO_TAG_ABRIDGED:
+            reader_clt = MTProtoCompactFrameStreamReader(reader_clt)
+            writer_clt = MTProtoCompactFrameStreamWriter(writer_clt)
+        elif proto_tag == PROTO_TAG_INTERMEDIATE:
+            reader_clt = MTProtoIntermediateFrameStreamReader(reader_clt)
+            writer_clt = MTProtoIntermediateFrameStreamWriter(writer_clt)
+        elif proto_tag == PROTO_TAG_SECURE:
+            reader_clt = MTProtoSecureIntermediateFrameStreamReader(reader_clt)
+            writer_clt = MTProtoSecureIntermediateFrameStreamWriter(writer_clt)
+        else:
+            return
+
+    tg_to_clt = tg_connect_reader_to_writer(reader_tg, writer_clt, user,
+                                            get_to_clt_bufsize(), False)
+    clt_to_tg = tg_connect_reader_to_writer(reader_clt, writer_tg,
+                                            user, get_to_tg_bufsize(), True)
+    task_tg_to_clt = asyncio.ensure_future(tg_to_clt)
+    task_clt_to_tg = asyncio.ensure_future(clt_to_tg)
+
+    update_user_stats(user, curr_connects=1)
+
+    tcp_limit_hit = (
+        user in config.USER_MAX_TCP_CONNS and
+        user_stats[user]["curr_connects"] > config.USER_MAX_TCP_CONNS[user]
+    )
+
+    user_expired = (
+        user in config.USER_EXPIRATIONS and
+        datetime.datetime.now() > config.USER_EXPIRATIONS[user]
+    )
+
+    user_data_quota_hit = (
+        user in config.USER_DATA_QUOTA and
+        (user_stats[user]["octets_to_client"] +
+         user_stats[user]["octets_from_client"] > config.USER_DATA_QUOTA[user])
+    )
+
+    if (not tcp_limit_hit) and (not user_expired) and (not user_data_quota_hit):
+        start = time.time()
+        await asyncio.wait([task_tg_to_clt, task_clt_to_tg], return_when=asyncio.FIRST_COMPLETED)
+        update_durations(time.time() - start)
+
+    update_user_stats(user, curr_connects=-1)
+
+    task_tg_to_clt.cancel()
+    task_clt_to_tg.cancel()
+
+    writer_tg.transport.abort()
+
+
+async def handle_client_wrapper(reader, writer):
+    try:
+        await handle_client(reader, writer)
+    except (asyncio.IncompleteReadError, asyncio.CancelledError):
+        pass
+    except (ConnectionResetError, TimeoutError, BrokenPipeError):
+        pass
+    except Exception:
+        traceback.print_exc()
+    finally:
+        writer.transport.abort()
+
+
+def make_metrics_pkt(metrics):
+    pkt_body_list = []
+    used_names = set()
+
+    for name, m_type, desc, val in metrics:
+        name = config.METRICS_PREFIX + name
+        if name not in used_names:
+            pkt_body_list.append("# HELP %s %s" % (name, desc))
+            pkt_body_list.append("# TYPE %s %s" % (name, m_type))
+            used_names.add(name)
+
+        if isinstance(val, dict):
+            tags = []
+            for tag, tag_val in val.items():
+                if tag == "val":
+                    continue
+                tag_val = tag_val.replace('"', r'\"')
+                tags.append('%s="%s"' % (tag, tag_val))
+            pkt_body_list.append("%s{%s} %s" % (name, ",".join(tags), val["val"]))
+        else:
+            pkt_body_list.append("%s %s" % (name, val))
+    pkt_body = "\n".join(pkt_body_list) + "\n"
+
+    pkt_header_list = []
+    pkt_header_list.append("HTTP/1.1 200 OK")
+    pkt_header_list.append("Connection: close")
+    pkt_header_list.append("Content-Length: %d" % len(pkt_body))
+    pkt_header_list.append("Content-Type: text/plain; version=0.0.4; charset=utf-8")
+    pkt_header_list.append("Date: %s" % time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime()))
+
+    pkt_header = "\r\n".join(pkt_header_list)
+
+    pkt = pkt_header + "\r\n\r\n" + pkt_body
+    return pkt
+
+
+async def handle_metrics(reader, writer):
+    global stats
+    global user_stats
+    global my_ip_info
+    global proxy_start_time
+    global proxy_links
+    global last_clients_with_time_skew
+    global last_clients_with_same_handshake
+
+    client_ip = writer.get_extra_info("peername")[0]
+    if client_ip not in config.METRICS_WHITELIST:
+        writer.close()
+        return
+
+    try:
+        metrics = []
+        metrics.append(["uptime", "counter", "proxy uptime", time.time() - proxy_start_time])
+        metrics.append(["connects_bad", "counter", "connects with bad secret",
+                       stats["connects_bad"]])
+        metrics.append(["connects_all", "counter", "incoming connects", stats["connects_all"]])
+        metrics.append(["handshake_timeouts", "counter", "number of timed out handshakes",
+                       stats["handshake_timeouts"]])
+
+        if config.METRICS_EXPORT_LINKS:
+            for link in proxy_links:
+                link_as_metric = link.copy()
+                link_as_metric["val"] = 1
+                metrics.append(["proxy_link_info", "counter",
+                                "the proxy link info", link_as_metric])
+
+        bucket_start = 0
+        for bucket in STAT_DURATION_BUCKETS:
+            bucket_end = bucket if bucket != STAT_DURATION_BUCKETS[-1] else "+Inf"
+            metric = {
+                "bucket": "%s-%s" % (bucket_start, bucket_end),
+                "val": stats["connects_with_duration_le_%s" % str(bucket)]
+            }
+            metrics.append(["connects_by_duration", "counter", "connects by duration", metric])
+            bucket_start = bucket_end
+
+        user_metrics_desc = [
+            ["user_connects", "counter", "user connects", "connects"],
+            ["user_connects_curr", "gauge", "current user connects", "curr_connects"],
+            ["user_octets", "counter", "octets proxied for user",
+                "octets_from_client+octets_to_client"],
+            ["user_msgs", "counter", "msgs proxied for user",
+                "msgs_from_client+msgs_to_client"],
+            ["user_octets_from", "counter", "octets proxied from user", "octets_from_client"],
+            ["user_octets_to", "counter", "octets proxied to user", "octets_to_client"],
+            ["user_msgs_from", "counter", "msgs proxied from user", "msgs_from_client"],
+            ["user_msgs_to", "counter", "msgs proxied to user", "msgs_to_client"],
+        ]
+
+        for m_name, m_type, m_desc, stat_key in user_metrics_desc:
+            for user, stat in user_stats.items():
+                if "+" in stat_key:
+                    val = 0
+                    for key_part in stat_key.split("+"):
+                        val += stat[key_part]
+                else:
+                    val = stat[stat_key]
+                metric = {"user": user, "val": val}
+                metrics.append([m_name, m_type, m_desc, metric])
+
+        pkt = make_metrics_pkt(metrics)
+        writer.write(pkt.encode())
+        await writer.drain()
+
+    except Exception:
+        traceback.print_exc()
+    finally:
+        writer.close()
+
+
+async def stats_printer():
+    global user_stats
+    global last_client_ips
+    global last_clients_with_time_skew
+    global last_clients_with_same_handshake
+
+    while True:
+        await asyncio.sleep(config.STATS_PRINT_PERIOD)
+
+        print("Stats for", time.strftime("%d.%m.%Y %H:%M:%S"))
+        for user, stat in user_stats.items():
+            print("%s: %d connects (%d current), %.2f MB, %d msgs" % (
+                user, stat["connects"], stat["curr_connects"],
+                (stat["octets_from_client"] + stat["octets_to_client"]) / 1000000,
+                stat["msgs_from_client"] + stat["msgs_to_client"]))
+        print(flush=True)
+
+        if last_client_ips:
+            print("New IPs:")
+            for ip in last_client_ips:
+                print(ip)
+            print(flush=True)
+            last_client_ips.clear()
+
+        if last_clients_with_time_skew:
+            print("Clients with time skew (possible replay-attackers):")
+            for ip, skew_minutes in last_clients_with_time_skew.items():
+                print("%s, clocks were %d minutes behind" % (ip, skew_minutes))
+            print(flush=True)
+            last_clients_with_time_skew.clear()
+        if last_clients_with_same_handshake:
+            print("Clients with duplicate handshake (likely replay-attackers):")
+            for ip, times in last_clients_with_same_handshake.items():
+                print("%s, %d times" % (ip, times))
+            print(flush=True)
+            last_clients_with_same_handshake.clear()
+
+
+async def make_https_req(url, host="core.telegram.org"):
+    """ Make request, return resp body and headers. """
+    SSL_PORT = 443
+    url_data = urllib.parse.urlparse(url)
+
+    HTTP_REQ_TEMPLATE = "\r\n".join(["GET %s HTTP/1.1", "Host: %s",
+                                     "Connection: close"]) + "\r\n\r\n"
+    reader, writer = await asyncio.open_connection(url_data.netloc, SSL_PORT, ssl=True)
+    req = HTTP_REQ_TEMPLATE % (urllib.parse.quote(url_data.path), host)
+    writer.write(req.encode("utf8"))
+    data = await reader.read()
+    writer.close()
+
+    headers, body = data.split(b"\r\n\r\n", 1)
+    return headers, body
+
+
+def gen_tls_client_hello_msg(server_name):
+    msg = bytearray()
+    msg += b"\x16\x03\x01\x02\x00\x01\x00\x01\xfc\x03\x03" + myrandom.getrandbytes(32)
+    msg += b"\x20" + myrandom.getrandbytes(32)
+    msg += b"\x00\x22\x4a\x4a\x13\x01\x13\x02\x13\x03\xc0\x2b\xc0\x2f\xc0\x2c\xc0\x30\xcc\xa9"
+    msg += b"\xcc\xa8\xc0\x13\xc0\x14\x00\x9c\x00\x9d\x00\x2f\x00\x35\x00\x0a\x01\x00\x01\x91"
+    msg += b"\xda\xda\x00\x00\x00\x00"
+    msg += int.to_bytes(len(server_name) + 5, 2, "big")
+    msg += int.to_bytes(len(server_name) + 3, 2, "big") + b"\x00"
+    msg += int.to_bytes(len(server_name), 2, "big") + server_name.encode("ascii")
+    msg += b"\x00\x17\x00\x00\xff\x01\x00\x01\x00\x00\x0a\x00\x0a\x00\x08\xaa\xaa\x00\x1d\x00"
+    msg += b"\x17\x00\x18\x00\x0b\x00\x02\x01\x00\x00\x23\x00\x00\x00\x10\x00\x0e\x00\x0c\x02"
+    msg += b"\x68\x32\x08\x68\x74\x74\x70\x2f\x31\x2e\x31\x00\x05\x00\x05\x01\x00\x00\x00\x00"
+    msg += b"\x00\x0d\x00\x14\x00\x12\x04\x03\x08\x04\x04\x01\x05\x03\x08\x05\x05\x01\x08\x06"
+    msg += b"\x06\x01\x02\x01\x00\x12\x00\x00\x00\x33\x00\x2b\x00\x29\xaa\xaa\x00\x01\x00\x00"
+    msg += b"\x1d\x00\x20" + gen_x25519_public_key()
+    msg += b"\x00\x2d\x00\x02\x01\x01\x00\x2b\x00\x0b\x0a\xba\xba\x03\x04\x03\x03\x03\x02\x03"
+    msg += b"\x01\x00\x1b\x00\x03\x02\x00\x02\x3a\x3a\x00\x01\x00\x00\x15"
+    msg += int.to_bytes(517 - len(msg) - 2, 2, "big")
+    msg += b"\x00" * (517 - len(msg))
+    return bytes(msg)
+
+
+async def get_encrypted_cert(host, port, server_name):
+    async def get_tls_record(reader):
+        try:
+            record_type = (await reader.readexactly(1))[0]
+            tls_version = await reader.readexactly(2)
+            if tls_version != b"\x03\x03":
+                return 0, b""
+            record_len = int.from_bytes(await reader.readexactly(2), "big")
+            record = await reader.readexactly(record_len)
+
+            return record_type, record
+        except asyncio.IncompleteReadError:
+            return 0, b""
+
+    reader, writer = await asyncio.open_connection(host, port)
+    writer.write(gen_tls_client_hello_msg(server_name))
+    await writer.drain()
+
+    record1_type, record1 = await get_tls_record(reader)
+    if record1_type != 22:
+        return b""
+
+    record2_type, record2 = await get_tls_record(reader)
+    if record2_type != 20:
+        return b""
+
+    record3_type, record3 = await get_tls_record(reader)
+    if record3_type != 23:
+        return b""
+
+    if len(record3) < MIN_CERT_LEN:
+        record4_type, record4 = await get_tls_record(reader)
+        if record4_type != 23:
+            return b""
+        msg = ("The MASK_HOST %s sent some TLS record before certificate record, this makes the " +
+               "proxy more detectable") % config.MASK_HOST
+        print_err(msg)
+
+        return record4
+
+    return record3
+
+
+async def get_mask_host_cert_len():
+    global fake_cert_len
+
+    GET_CERT_TIMEOUT = 10
+    MASK_ENABLING_CHECK_PERIOD = 60
+
+    while True:
+        try:
+            if not config.MASK:
+                # do nothing
+                await asyncio.sleep(MASK_ENABLING_CHECK_PERIOD)
+                continue
+
+            task = get_encrypted_cert(config.MASK_HOST, config.MASK_PORT, config.TLS_DOMAIN)
+            cert = await asyncio.wait_for(task, timeout=GET_CERT_TIMEOUT)
+            if cert:
+                if len(cert) < MIN_CERT_LEN:
+                    msg = ("The MASK_HOST %s returned several TLS records, this is not supported" %
+                           config.MASK_HOST)
+                    print_err(msg)
+                elif len(cert) != fake_cert_len:
+                    fake_cert_len = len(cert)
+                    print_err("Got cert from the MASK_HOST %s, its length is %d" %
+                              (config.MASK_HOST, fake_cert_len))
+            else:
+                print_err("The MASK_HOST %s is not TLS 1.3 host, this is not recommended" %
+                          config.MASK_HOST)
+        except ConnectionRefusedError:
+            print_err("The MASK_HOST %s is refusing connections, this is not recommended" %
+                      config.MASK_HOST)
+        except (TimeoutError, asyncio.TimeoutError):
+            print_err("Got timeout while getting TLS handshake from MASK_HOST %s" %
+                      config.MASK_HOST)
+        except Exception as E:
+            print_err("Failed to connect to MASK_HOST %s: %s" % (
+                      config.MASK_HOST, E))
+
+        await asyncio.sleep(config.GET_CERT_LEN_PERIOD)
+
+
+async def get_srv_time():
+    TIME_SYNC_ADDR = "https://core.telegram.org/getProxySecret"
+    MAX_TIME_SKEW = 30
+
+    global disable_middle_proxy
+    global is_time_skewed
+
+    want_to_reenable_advertising = False
+    while True:
+        try:
+            headers, secret = await make_https_req(TIME_SYNC_ADDR)
+
+            for line in headers.split(b"\r\n"):
+                if not line.startswith(b"Date: "):
+                    continue
+                line = line[len("Date: "):].decode()
+                srv_time = datetime.datetime.strptime(line, "%a, %d %b %Y %H:%M:%S %Z")
+                srv_time = srv_time.replace(tzinfo=datetime.timezone.utc)
+                now_time = datetime.datetime.now(datetime.timezone.utc)
+                is_time_skewed = (now_time-srv_time).total_seconds() > MAX_TIME_SKEW
+                if is_time_skewed and config.USE_MIDDLE_PROXY and not disable_middle_proxy:
+                    print_err("Time skew detected, please set the clock")
+                    print_err("Server time:", srv_time, "your time:", now_time)
+                    print_err("Disabling advertising to continue serving")
+                    print_err("Putting down the shields against replay attacks")
+
+                    disable_middle_proxy = True
+                    want_to_reenable_advertising = True
+                elif not is_time_skewed and want_to_reenable_advertising:
+                    print_err("Time is ok, reenabling advertising")
+                    disable_middle_proxy = False
+                    want_to_reenable_advertising = False
+        except Exception as E:
+            print_err("Error getting server time", E)
+
+        await asyncio.sleep(config.GET_TIME_PERIOD)
+
+
+async def clear_ip_resolving_cache():
+    global mask_host_cached_ip
+    min_sleep = myrandom.randrange(60 - 10, 60 + 10)
+    max_sleep = myrandom.randrange(120 - 10, 120 + 10)
+    while True:
+        mask_host_cached_ip = None
+        await asyncio.sleep(myrandom.randrange(min_sleep, max_sleep))
+
+
+async def update_middle_proxy_info():
+    async def get_new_proxies(url):
+        PROXY_REGEXP = re.compile(r"proxy_for\s+(-?\d+)\s+(.+):(\d+)\s*;")
+        ans = {}
+        headers, body = await make_https_req(url)
+
+        fields = PROXY_REGEXP.findall(body.decode("utf8"))
+        if fields:
+            for dc_idx, host, port in fields:
+                if host.startswith("[") and host.endswith("]"):
+                    host = host[1:-1]
+                dc_idx, port = int(dc_idx), int(port)
+                if dc_idx not in ans:
+                    ans[dc_idx] = [(host, port)]
+                else:
+                    ans[dc_idx].append((host, port))
+        return ans
+
+    PROXY_INFO_ADDR = "https://core.telegram.org/getProxyConfig"
+    PROXY_INFO_ADDR_V6 = "https://core.telegram.org/getProxyConfigV6"
+    PROXY_SECRET_ADDR = "https://core.telegram.org/getProxySecret"
+
+    global TG_MIDDLE_PROXIES_V4
+    global TG_MIDDLE_PROXIES_V6
+    global PROXY_SECRET
+
+    while True:
+        try:
+            v4_proxies = await get_new_proxies(PROXY_INFO_ADDR)
+            if not v4_proxies:
+                raise Exception("no proxy data")
+            TG_MIDDLE_PROXIES_V4 = v4_proxies
+        except Exception as E:
+            print_err("Error updating middle proxy list:", E)
+
+        try:
+            v6_proxies = await get_new_proxies(PROXY_INFO_ADDR_V6)
+            if not v6_proxies:
+                raise Exception("no proxy data (ipv6)")
+            TG_MIDDLE_PROXIES_V6 = v6_proxies
+        except Exception as E:
+            print_err("Error updating middle proxy list for IPv6:", E)
+
+        try:
+            headers, secret = await make_https_req(PROXY_SECRET_ADDR)
+            if not secret:
+                raise Exception("no secret")
+            if secret != PROXY_SECRET:
+                PROXY_SECRET = secret
+                print_err("Middle proxy secret updated")
+        except Exception as E:
+            print_err("Error updating middle proxy secret, using old", E)
+
+        await asyncio.sleep(config.PROXY_INFO_UPDATE_PERIOD)
+
+
+def init_ip_info():
+    global my_ip_info
+    global disable_middle_proxy
+
+    def get_ip_from_url(url):
+        TIMEOUT = 5
+        try:
+            with urllib.request.urlopen(url, timeout=TIMEOUT) as f:
+                if f.status != 200:
+                    raise Exception("Invalid status code")
+                return f.read().decode().strip()
+        except Exception:
+            return None
+
+    IPV4_URL1 = "http://v4.ident.me/"
+    IPV4_URL2 = "http://ipv4.icanhazip.com/"
+
+    IPV6_URL1 = "http://v6.ident.me/"
+    IPV6_URL2 = "http://ipv6.icanhazip.com/"
+
+    my_ip_info["ipv4"] = get_ip_from_url(IPV4_URL1) or get_ip_from_url(IPV4_URL2)
+    my_ip_info["ipv6"] = get_ip_from_url(IPV6_URL1) or get_ip_from_url(IPV6_URL2)
+
+    # the server can return ipv4 address instead of ipv6
+    if my_ip_info["ipv6"] and ":" not in my_ip_info["ipv6"]:
+        my_ip_info["ipv6"] = None
+
+    if my_ip_info["ipv6"] and (config.PREFER_IPV6 or not my_ip_info["ipv4"]):
+        print_err("IPv6 found, using it for external communication")
+
+    if config.USE_MIDDLE_PROXY:
+        if not my_ip_info["ipv4"] and not my_ip_info["ipv6"]:
+            print_err("Failed to determine your ip, advertising disabled")
+            disable_middle_proxy = True
+
+
+def print_tg_info():
+    global my_ip_info
+    global proxy_links
+
+    print_default_warning = False
+
+    if config.PORT == 3256:
+        print("The default port 3256 is used, this is not recommended", flush=True)
+        if not config.MODES["classic"] and not config.MODES["secure"]:
+            print("Since you have TLS only mode enabled the best port is 443", flush=True)
+        print_default_warning = True
+
+    if not config.MY_DOMAIN:
+        ip_addrs = [ip for ip in my_ip_info.values() if ip]
+        if not ip_addrs:
+            ip_addrs = ["YOUR_IP"]
+    else:
+        ip_addrs = [config.MY_DOMAIN]
+
+    proxy_links = []
+
+    for user, secret in sorted(config.USERS.items(), key=lambda x: x[0]):
+        for ip in ip_addrs:
+            if config.MODES["classic"]:
+                params = {"server": ip, "port": config.PORT, "secret": secret}
+                params_encoded = urllib.parse.urlencode(params, safe=':')
+                classic_link = "tg://proxy?{}".format(params_encoded)
+                proxy_links.append({"user": user, "link": classic_link})
+                print("{}: {}".format(user, classic_link), flush=True)
+
+            if config.MODES["secure"]:
+                params = {"server": ip, "port": config.PORT, "secret": "dd" + secret}
+                params_encoded = urllib.parse.urlencode(params, safe=':')
+                dd_link = "tg://proxy?{}".format(params_encoded)
+                proxy_links.append({"user": user, "link": dd_link})
+                print("{}: {}".format(user, dd_link), flush=True)
+
+            if config.MODES["tls"]:
+                tls_secret = "ee" + secret + config.TLS_DOMAIN.encode().hex()
+                # the base64 links is buggy on ios
+                # tls_secret = bytes.fromhex("ee" + secret) + config.TLS_DOMAIN.encode()
+                # tls_secret_base64 = base64.urlsafe_b64encode(tls_secret)
+                params = {"server": ip, "port": config.PORT, "secret": tls_secret}
+                params_encoded = urllib.parse.urlencode(params, safe=':')
+                tls_link = "tg://proxy?{}".format(params_encoded)
+                proxy_links.append({"user": user, "link": tls_link})
+                print("{}: {}".format(user, tls_link), flush=True)
+
+        if secret in ["00000000000000000000000000000000", "0123456789abcdef0123456789abcdef",
+                      "00000000000000000000000000000001"]:
+            msg = "The default secret {} is used, this is not recommended".format(secret)
+            print(msg, flush=True)
+            random_secret = "".join(myrandom.choice("0123456789abcdef") for i in range(32))
+            print("You can change it to this random secret:", random_secret, flush=True)
+            print_default_warning = True
+
+    if config.TLS_DOMAIN == "www.google.com":
+        print("The default TLS_DOMAIN www.google.com is used, this is not recommended", flush=True)
+        msg = "You should use random existing domain instead, bad clients are proxied there"
+        print(msg, flush=True)
+        print_default_warning = True
+
+    if print_default_warning:
+        print_err("Warning: one or more default settings detected")
+
+
+def setup_files_limit():
+    try:
+        import resource
+        soft_fd_limit, hard_fd_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
+        resource.setrlimit(resource.RLIMIT_NOFILE, (hard_fd_limit, hard_fd_limit))
+    except (ValueError, OSError):
+        print("Failed to increase the limit of opened files", flush=True, file=sys.stderr)
+    except ImportError:
+        pass
+
+
+def setup_asyncio():
+    # get rid of annoying "socket.send() raised exception" log messages
+    asyncio.constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES = 100
+
+
+def setup_signals():
+    if hasattr(signal, 'SIGUSR1'):
+        def debug_signal(signum, frame):
+            import pdb
+            pdb.set_trace()
+
+        signal.signal(signal.SIGUSR1, debug_signal)
+
+    if hasattr(signal, 'SIGUSR2'):
+        def reload_signal(signum, frame):
+            init_config()
+            ensure_users_in_user_stats()
+            apply_upstream_proxy_settings()
+            print("Config reloaded", flush=True, file=sys.stderr)
+            print_tg_info()
+
+        signal.signal(signal.SIGUSR2, reload_signal)
+
+
+def try_setup_uvloop():
+    if config.SOCKS5_HOST and config.SOCKS5_PORT:
+        # socks mode is not compatible with uvloop
+        return
+    try:
+        import uvloop
+        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+        print_err("Found uvloop, using it for optimal performance")
+    except ImportError:
+        pass
+
+
+def remove_unix_socket(path):
+    try:
+        if stat.S_ISSOCK(os.stat(path).st_mode):
+            os.unlink(path)
+    except (FileNotFoundError, NotADirectoryError):
+        pass
+
+
+def loop_exception_handler(loop, context):
+    exception = context.get("exception")
+    transport = context.get("transport")
+    if exception:
+        if isinstance(exception, TimeoutError):
+            if transport:
+                transport.abort()
+                return
+        if isinstance(exception, OSError):
+            IGNORE_ERRNO = {
+                10038,  # operation on non-socket on Windows, likely because fd == -1
+                121,    # the semaphore timeout period has expired on Windows
+            }
+
+            FORCE_CLOSE_ERRNO = {
+                113,    # no route to host
+
+            }
+            if exception.errno in IGNORE_ERRNO:
+                return
+            elif exception.errno in FORCE_CLOSE_ERRNO:
+                if transport:
+                    transport.abort()
+                    return
+
+    loop.default_exception_handler(context)
+
+
+def create_servers(loop):
+    servers = []
+
+    reuse_port = hasattr(socket, "SO_REUSEPORT")
+    has_unix = hasattr(socket, "AF_UNIX")
+
+    if config.LISTEN_ADDR_IPV4:
+        task = asyncio.start_server(handle_client_wrapper, config.LISTEN_ADDR_IPV4, config.PORT,
+                                    limit=get_to_tg_bufsize(), reuse_port=reuse_port)
+        servers.append(loop.run_until_complete(task))
+
+    if config.LISTEN_ADDR_IPV6 and socket.has_ipv6:
+        task = asyncio.start_server(handle_client_wrapper, config.LISTEN_ADDR_IPV6, config.PORT,
+                                    limit=get_to_tg_bufsize(), reuse_port=reuse_port)
+        servers.append(loop.run_until_complete(task))
+
+    if config.LISTEN_UNIX_SOCK and has_unix:
+        remove_unix_socket(config.LISTEN_UNIX_SOCK)
+        task = asyncio.start_unix_server(handle_client_wrapper, config.LISTEN_UNIX_SOCK,
+                                         limit=get_to_tg_bufsize())
+        servers.append(loop.run_until_complete(task))
+        os.chmod(config.LISTEN_UNIX_SOCK, 0o666)
+
+    if config.METRICS_PORT is not None:
+        if config.METRICS_LISTEN_ADDR_IPV4:
+            task = asyncio.start_server(handle_metrics, config.METRICS_LISTEN_ADDR_IPV4,
+                                        config.METRICS_PORT)
+            servers.append(loop.run_until_complete(task))
+        if config.METRICS_LISTEN_ADDR_IPV6 and socket.has_ipv6:
+            task = asyncio.start_server(handle_metrics, config.METRICS_LISTEN_ADDR_IPV6,
+                                        config.METRICS_PORT)
+            servers.append(loop.run_until_complete(task))
+
+    return servers
+
+
+def create_utilitary_tasks(loop):
+    tasks = []
+
+    stats_printer_task = asyncio.Task(stats_printer(), loop=loop)
+    tasks.append(stats_printer_task)
+
+    if config.USE_MIDDLE_PROXY:
+        middle_proxy_updater_task = asyncio.Task(update_middle_proxy_info(), loop=loop)
+        tasks.append(middle_proxy_updater_task)
+
+        if config.GET_TIME_PERIOD:
+            time_get_task = asyncio.Task(get_srv_time(), loop=loop)
+            tasks.append(time_get_task)
+
+    get_cert_len_task = asyncio.Task(get_mask_host_cert_len(), loop=loop)
+    tasks.append(get_cert_len_task)
+
+    clear_resolving_cache_task = asyncio.Task(clear_ip_resolving_cache(), loop=loop)
+    tasks.append(clear_resolving_cache_task)
+
+    return tasks
+
+
+def main():
+    init_config()
+    ensure_users_in_user_stats()
+    apply_upstream_proxy_settings()
+    init_ip_info()
+    print_tg_info()
+
+    setup_asyncio()
+    setup_files_limit()
+    setup_signals()
+    try_setup_uvloop()
+
+    init_proxy_start_time()
+
+    if sys.platform == "win32":
+        loop = asyncio.ProactorEventLoop()
+    else:
+        loop = asyncio.new_event_loop()
+
+    asyncio.set_event_loop(loop)
+    loop.set_exception_handler(loop_exception_handler)
+
+    utilitary_tasks = create_utilitary_tasks(loop)
+    for task in utilitary_tasks:
+        asyncio.ensure_future(task)
+
+    servers = create_servers(loop)
+
+    try:
+        loop.run_forever()
+    except KeyboardInterrupt:
+        pass
+
+    if hasattr(asyncio, "all_tasks"):
+        tasks = asyncio.all_tasks(loop)
+    else:
+        # for compatibility with Python 3.6
+        tasks = asyncio.Task.all_tasks(loop)
+
+    for task in tasks:
+        task.cancel()
+
+    for server in servers:
+        server.close()
+        loop.run_until_complete(server.wait_closed())
+
+    has_unix = hasattr(socket, "AF_UNIX")
+
+    if config.LISTEN_UNIX_SOCK and has_unix:
+        remove_unix_socket(config.LISTEN_UNIX_SOCK)
+
+    loop.close()
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,4 @@
+cryptography==42.0.5
+uvloop==0.19.0
+flask==3.0.3
+psutil==5.9.8

runner.py

@@ -0,0 +1,37 @@
+import threading
+import sys
+import os
+
+# Set uvloop policy if available
+try:
+    import uvloop
+    import asyncio
+    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+    print("Using uvloop for maximum performance.", flush=True)
+except ImportError:
+    print("uvloop not installed, using default asyncio loop.", flush=True)
+
+import mtprotoproxy
+from webui import start_webui
+
+def main():
+    # Start the Flask WebUI in a background daemon thread
+    web_thread = threading.Thread(target=start_webui, daemon=True)
+    web_thread.start()
+    print("Started WebUI on port 7860.", flush=True)
+
+    # We need to simulate arguments to mtprotoproxy if it expects sys.argv[1] as config
+    if len(sys.argv) < 2:
+        # Defaulting sys.argv to have config.py as first argument so mtprotoproxy picks it up
+        sys.argv.append("config.py")
+
+    print("Starting MTProto Proxy...", flush=True)
+    # Start mtprotoproxy main loop
+    # mtprotoproxy.main() starts the loop and blocks forever
+    try:
+        mtprotoproxy.main()
+    except KeyboardInterrupt:
+        print("Interrupted by user. Exiting...", flush=True)
+
+if __name__ == "__main__":
+    main()

test_docker.sh

@@ -0,0 +1 @@
+docker run --rm python:3.12-slim bash -c "apt-get update && apt-get install -y gcc python3-dev && pip install uvloop cryptography flask psutil gunicorn"

webui.py

@@ -0,0 +1,137 @@
+from flask import Flask, jsonify, render_template_string
+import psutil
+import mtprotoproxy
+import time
+import os
+
+app = Flask(__name__)
+
+HTML_TEMPLATE = """
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>MTProto Proxy Dashboard</title>
+    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
+    <style>
+        body { background-color: #f8f9fa; padding-top: 20px; }
+        .card { margin-bottom: 20px; box-shadow: 0 4px 6px rgba(0,0,0,0.1); }
+        .metric-value { font-size: 2rem; font-weight: bold; color: #0d6efd; }
+    </style>
+    <script>
+        async function updateStats() {
+            try {
+                const response = await fetch('/api/stats');
+                const data = await response.json();
+
+                document.getElementById('cpu-load').innerText = data.cpu_percent + '%';
+                document.getElementById('mem-load').innerText = data.mem_percent + '%';
+
+                // Calculate total connections
+                let totalConnections = 0;
+                let userHtml = '';
+                for (const [user, stats] of Object.entries(data.user_stats)) {
+                    totalConnections += stats.curr_connects || 0;
+                    userHtml += `<tr>
+                        <td>${user}</td>
+                        <td>${stats.curr_connects || 0}</td>
+                        <td>${stats.connects || 0}</td>
+                        <td>${(((stats.octets_from_client || 0) + (stats.octets_to_client || 0)) / 1024 / 1024).toFixed(2)} MB</td>
+                    </tr>`;
+                }
+                document.getElementById('total-users').innerText = totalConnections;
+                document.getElementById('user-table-body').innerHTML = userHtml;
+            } catch (e) {
+                console.error("Failed to fetch stats", e);
+            }
+        }
+
+        setInterval(updateStats, 2000);
+        window.onload = updateStats;
+    </script>
+</head>
+<body>
+    <div class="container">
+        <h1 class="mb-4 text-center">MTProto Proxy Dashboard</h1>
+
+        <div class="row">
+            <div class="col-md-4">
+                <div class="card text-center">
+                    <div class="card-body">
+                        <h5 class="card-title">Active Connections</h5>
+                        <p class="metric-value" id="total-users">0</p>
+                    </div>
+                </div>
+            </div>
+            <div class="col-md-4">
+                <div class="card text-center">
+                    <div class="card-body">
+                        <h5 class="card-title">CPU Load</h5>
+                        <p class="metric-value" id="cpu-load">0%</p>
+                    </div>
+                </div>
+            </div>
+            <div class="col-md-4">
+                <div class="card text-center">
+                    <div class="card-body">
+                        <h5 class="card-title">Memory Load</h5>
+                        <p class="metric-value" id="mem-load">0%</p>
+                    </div>
+                </div>
+            </div>
+        </div>
+
+        <div class="card mt-4">
+            <div class="card-header bg-primary text-white">
+                <h5 class="mb-0">User Statistics</h5>
+            </div>
+            <div class="card-body">
+                <table class="table table-striped">
+                    <thead>
+                        <tr>
+                            <th>User Name</th>
+                            <th>Active Connections</th>
+                            <th>Total Connects</th>
+                            <th>Data Transferred</th>
+                        </tr>
+                    </thead>
+                    <tbody id="user-table-body">
+                        <!-- Filled by JS -->
+                    </tbody>
+                </table>
+            </div>
+        </div>
+    </div>
+</body>
+</html>
+"""
+
+@app.route('/')
+def index():
+    return render_template_string(HTML_TEMPLATE)
+
+@app.route('/api/stats')
+def stats():
+    user_stats = {}
+    try:
+        # mtprotoproxy.user_stats is a defaultdict of Counter
+        # Need to copy list of items first since dictionary can change size during iteration
+        items = list(mtprotoproxy.user_stats.items())
+        for user, counter in items:
+            user_stats[user] = dict(counter)
+    except Exception as e:
+        print(f"Error reading stats: {e}")
+
+    return jsonify({
+        "cpu_percent": psutil.cpu_percent(interval=None),
+        "mem_percent": psutil.virtual_memory().percent,
+        "user_stats": user_stats
+    })
+
+def start_webui():
+    # Make sure we don't start the reloader, as we're running alongside asyncio loop
+    app.run(host='0.0.0.0', port=7860, debug=False, use_reloader=False)
+
+if __name__ == "__main__":
+    start_webui()