abcd

Surendra9123 · Surendra9123 · commit 8799631d679e · 2026-06-27T12:45:51.000+05:30
diff --git a/pyrogram/connection/connection.py b/pyrogram/connection/connection.py
@@ -22,7 +22,7 @@
 
 from pyrogram import utils
 
-from .transport import TCP, TCPAbridged
+from .transport import TCP, TCPAbridged, TCPMTProxy
 
 log = logging.getLogger(__name__)
 
@@ -51,6 +51,12 @@ def __init__(
         self.media = media
         self.protocol_factory = protocol_factory
         self.crypto_executor_workers = crypto_executor_workers
+        self.is_mtproxy = False
+
+        # if isinstance(proxy, str) and proxy.lower().startswith("mtproxy://"):
+        #     self.is_mtproxy = True
+        if isinstance(proxy, dict) and proxy.get("scheme", "").lower() == "mtproxy":
+            self.is_mtproxy = True
 
         self.protocol: Optional[TCP] = None
 
@@ -61,8 +67,10 @@ def __init__(
 
     async def connect(self) -> None:
         for i in range(Connection.MAX_CONNECTION_ATTEMPTS):
-            self.protocol = self.protocol_factory(ipv6=self.ipv6, proxy=self.proxy, crypto_executor_workers=self.crypto_executor_workers, loop=self.loop)
-
+            if self.is_mtproxy:
+                self.protocol = TCPMTProxy(dc_id=self.dc_id, ipv6=self.ipv6, proxy=self.proxy, crypto_executor_workers=self.crypto_executor_workers, loop=self.loop)
+            else:
+                self.protocol = self.protocol_factory(ipv6=self.ipv6, proxy=self.proxy, crypto_executor_workers=self.crypto_executor_workers, loop=self.loop)
             try:
                 log.info("Connecting...")
                 await self.protocol.connect((self.server_address, self.port))
diff --git a/pyrogram/connection/transport/tcp/__init__.py b/pyrogram/connection/transport/tcp/__init__.py
@@ -22,3 +22,4 @@
 from .tcp_full import TCPFull
 from .tcp_intermediate import TCPIntermediate
 from .tcp_intermediate_o import TCPIntermediateO
+from .tcp_mtproxy import TCPMTProxy
diff --git a/pyrogram/connection/transport/tcp/faketls.py b/pyrogram/connection/transport/tcp/faketls.py
@@ -0,0 +1,222 @@
+import os
+import struct
+import random
+
+# Faithful Python port of Telegram tgnet's TlsHello (ConnectionSocket.cpp).
+# Produces a modern Chrome-like ClientHello (GREASE, TLS 1.3 cipher suites,
+# permuted extensions, X25519 + ML-KEM-768 hybrid key_share, ECH, ALPN, padding).
+# The 32-byte client-random field at offset [11:43] is left zeroed; the caller
+# fills it with HMAC(secret, hello)[:28] + (HMAC[28:32] XOR timestamp).
+
+MAX_GREASE = 8
+P25519 = 2 ** 255 - 19
+
+
+# ---- curve25519 public-key generation (matches generate_public_key) ----
+
+def _get_y2(x, mod):
+    # y^2 = x^3 + 486662*x^2 + x
+    y = (x + 486662) % mod
+    y = (y * x) % mod
+    y = (y + 1) % mod
+    y = (y * x) % mod
+    return y
+
+
+def _get_double_x(x, mod):
+    # x_2 = (x^2 - 1)^2 / (4 * y^2)
+    denominator = (_get_y2(x, mod) * 4) % mod
+    numerator = (x * x - 1) % mod
+    numerator = (numerator * numerator) % mod
+    denominator_inv = pow(denominator, mod - 2, mod)
+    return (numerator * denominator_inv) % mod
+
+
+def generate_public_key():
+    mod = P25519
+    pw = (mod - 1) // 2  # 2^254 - 10
+    while True:
+        key = bytearray(os.urandom(32))
+        key[31] &= 127
+        x = int.from_bytes(key, "big")
+        x = (x * x) % mod
+        if pow(_get_y2(x, mod), pw, mod) == 1:
+            break
+    for _ in range(3):
+        x = _get_double_x(x, mod)
+    return x.to_bytes(32, "big")[::-1]
+
+
+# ---- ML-KEM-768 fake key generation (matches generate_key_ml_kem_768) ----
+
+def generate_key_ml_kem_768():
+    Q = 3329
+    N = 384
+    key = bytearray(1184)
+    values = struct.unpack("<%dI" % (N * 2), os.urandom(N * 2 * 4))
+    for i in range(N):
+        a = values[i * 2] % Q
+        b = values[i * 2 + 1] % Q
+        key[i * 3 + 0] = a & 0xFF
+        key[i * 3 + 1] = ((a >> 8) | ((b & 0x0F) << 4)) & 0xFF
+        key[i * 3 + 2] = (b >> 4) & 0xFF
+    key[1152:1184] = os.urandom(32)
+    return bytes(key)
+
+
+# ---- GREASE values (matches TlsHello constructor) ----
+
+def _gen_grease():
+    g = bytearray(os.urandom(MAX_GREASE))
+    for a in range(MAX_GREASE):
+        g[a] = (g[a] & 0xf0) + 0x0A
+    for i in range(1, MAX_GREASE, 2):
+        if i + 1 < MAX_GREASE and g[i] == g[i + 1]:
+            g[i] ^= 0x10
+    return g
+
+
+# ---- op interpreter (matches writeOp) ----
+# Op forms (tuples):
+#   ('str', b'...')      ('rand', n)    ('K',)    ('M',)    ('E',)
+#   ('zero', n)          ('domain',)    ('grease', seed)
+#   ('begin',)           ('end',)       ('P',)    ('perm', [[ops], ...])
+
+def _write_op(op, out, scopes, grease, domain):
+    t = op[0]
+    if t == 'str':
+        out += op[1]
+    elif t == 'rand':
+        out += os.urandom(op[1])
+    elif t == 'K':
+        out += generate_public_key()
+    elif t == 'M':
+        out += generate_key_ml_kem_768()
+    elif t == 'zero':
+        out += b"\x00" * op[1]
+    elif t == 'domain':
+        d = domain.encode("ascii", "ignore")[:253]
+        out += d
+    elif t == 'grease':
+        g = grease[op[1]]
+        out += bytes([g, g])
+    elif t == 'begin':
+        scopes.append(len(out))
+        out += b"\x00\x00"
+    elif t == 'end':
+        begin = scopes.pop()
+        size = len(out) - begin - 2
+        out[begin] = (size >> 8) & 0xFF
+        out[begin + 1] = size & 0xFF
+    elif t == 'E':
+        length = (144, 176, 208, 240)[random.randrange(4)]
+        out += os.urandom(length)
+    elif t == 'P':
+        length = len(out)
+        if length <= 513:
+            _write_op(('str', b"\x00\x15"), out, scopes, grease, domain)
+            _write_op(('begin',), out, scopes, grease, domain)
+            _write_op(('zero', 513 - length), out, scopes, grease, domain)
+            _write_op(('end',), out, scopes, grease, domain)
+    elif t == 'perm':
+        parts = list(op[1])
+        n = len(parts)
+        for i in range(n - 1):
+            j = i + random.randrange(n - i)
+            if i != j:
+                parts[i], parts[j] = parts[j], parts[i]
+        for part in parts:
+            for o in part:
+                _write_op(o, out, scopes, grease, domain)
+    else:
+        raise ValueError("unknown op %r" % (t,))
+
+
+def _default_ops():
+    s = lambda b: ('str', b)
+    return [
+        s(b"\x16\x03\x01"),
+        ('begin',),
+        s(b"\x01\x00"),
+        ('begin',),
+        s(b"\x03\x03"),
+        ('zero', 32),
+        s(b"\x20"),
+        ('rand', 32),
+        s(b"\x00\x20"),
+        ('grease', 0),
+        s(b"\x13\x01\x13\x02\x13\x03\xc0\x2b\xc0\x2f\xc0\x2c\xc0\x30\xcc\xa9\xcc\xa8\xc0\x13\xc0\x14\x00\x9c\x00\x9d\x00\x2f\x00\x35\x01\x00"),
+        ('begin',),
+        ('grease', 2),
+        s(b"\x00\x00"),
+        ('perm', [
+            [
+                s(b"\x00\x00"),
+                ('begin',),
+                ('begin',),
+                s(b"\x00"),
+                ('begin',),
+                ('domain',),
+                ('end',),
+                ('end',),
+                ('end',),
+            ],
+            [s(b"\x00\x05\x00\x05\x01\x00\x00\x00\x00")],
+            [
+                s(b"\x00\x0a\x00\x0c\x00\x0a"),
+                ('grease', 4),
+                s(b"\x11\xec\x00\x1d\x00\x17\x00\x18"),
+            ],
+            [s(b"\x00\x0b\x00\x02\x01\x00")],
+            [s(b"\x00\x0d\x00\x12\x00\x10\x04\x03\x08\x04\x04\x01\x05\x03\x08\x05\x05\x01\x08\x06\x06\x01")],
+            [s(b"\x00\x10\x00\x0e\x00\x0c\x02\x68\x32\x08\x68\x74\x74\x70\x2f\x31\x2e\x31")],
+            [s(b"\x00\x12\x00\x00")],
+            [s(b"\x00\x17\x00\x00")],
+            [s(b"\x00\x1b\x00\x03\x02\x00\x02")],
+            [s(b"\x00\x23\x00\x00")],
+            [
+                s(b"\x00\x2b\x00\x07\x06"),
+                ('grease', 6),
+                s(b"\x03\x04\x03\x03"),
+            ],
+            [s(b"\x00\x2d\x00\x02\x01\x01")],
+            [
+                s(b"\x00\x33\x04\xef\x04\xed"),
+                ('grease', 4),
+                s(b"\x00\x01\x00\x11\xec\x04\xc0"),
+                ('M',),
+                ('K',),
+                s(b"\x00\x1d\x00\x20"),
+                ('K',),
+            ],
+            [s(b"\x44\xcd\x00\x05\x00\x03\x02\x68\x32")],
+            [
+                s(b"\xfe\x0d"),
+                ('begin',),
+                s(b"\x00\x00\x01\x00\x01"),
+                ('rand', 1),
+                s(b"\x00\x20"),
+                ('rand', 32),
+                ('begin',),
+                ('E',),
+                ('end',),
+                ('end',),
+            ],
+            [s(b"\xff\x01\x00\x01\x00")],
+        ]),
+        ('grease', 3),
+        s(b"\x00\x01\x00"),
+        ('P',),
+        ('end',),
+        ('end',),
+        ('end',),
+    ]
+
+
+def build_fake_tls_client_hello(domain: str) -> bytearray:
+    grease = _gen_grease()
+    out = bytearray()
+    scopes = []
+    for op in _default_ops():
+        _write_op(op, out, scopes, grease, domain)
+    return out
diff --git a/pyrogram/connection/transport/tcp/tcp.py b/pyrogram/connection/transport/tcp/tcp.py
@@ -36,6 +36,7 @@ class ProxyDict(TypedDict):
     scheme: str
     hostname: str
     port: int
+    secret: Optional[str]
     username: Optional[str]
     password: Optional[str]
 
@@ -154,7 +155,7 @@ async def _connect_via_direct(self, destination: Tuple[str, int]) -> None:
         log.info("Connection established")
 
     async def _connect(self, destination: Tuple[str, int]) -> None:
-        if self.proxy:
+        if self.proxy and isinstance(self.proxy, dict) and self.proxy.get("scheme", "").lower() != "mtproxy":
             await self._connect_via_proxy(destination)
         else:
             await self._connect_via_direct(destination)
@@ -245,4 +246,4 @@ async def recv(self, length: int = 0) -> Optional[bytes]:
                     return None
 
         log.debug("Recv complete: %d bytes", len(data))
-        return data
+        return data
diff --git a/pyrogram/connection/transport/tcp/tcp_mtproxy.py b/pyrogram/connection/transport/tcp/tcp_mtproxy.py
diff --git a/pyrogram/session/session.py b/pyrogram/session/session.py
