mirror of
https://github.com/alexbers/mtprotoproxy.git
synced 2026-03-19 17:15:50 +00:00
Compare commits
31 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
027ee2713b | ||
|
|
3c4c92cdb2 | ||
|
|
3ef826cd6b | ||
|
|
9ec1e543bd | ||
|
|
fae04ed3c7 | ||
|
|
25685f370c | ||
|
|
1feb8e5fc7 | ||
|
|
5e488203a2 | ||
|
|
94fd98a1fb | ||
|
|
4c5b0803fa | ||
|
|
8fa8aabc8e | ||
|
|
f2fbaa923b | ||
|
|
79eaabdd23 | ||
|
|
23c7b0d53b | ||
|
|
516600a32d | ||
|
|
5fcd1c0158 | ||
|
|
781549f37f | ||
|
|
dc1223fd90 | ||
|
|
0d52ae0bc7 | ||
|
|
2b1469985d | ||
|
|
4784491800 | ||
|
|
7a2c6b9825 | ||
|
|
37307a98fb | ||
|
|
02b39168c3 | ||
|
|
163e7b7cce | ||
|
|
1ed13d9efa | ||
|
|
51c8d68271 | ||
|
|
50cd74051f | ||
|
|
1d826866d1 | ||
|
|
ff6b826e13 | ||
|
|
3315ac1df6 |
@@ -1,6 +1,6 @@
|
|||||||
# Async MTProto Proxy #
|
# Async MTProto Proxy #
|
||||||
|
|
||||||
Fast and simple to setup mtproto proxy written on Python.
|
Fast and simple to setup MTProto proxy written in Python.
|
||||||
|
|
||||||
## Starting Up ##
|
## Starting Up ##
|
||||||
|
|
||||||
@@ -26,4 +26,4 @@ The proxy can be launched:
|
|||||||
- with a custom config: `python3 mtprotoproxy.py [configfile]`
|
- with a custom config: `python3 mtprotoproxy.py [configfile]`
|
||||||
- several times, clients will be automaticaly balanced between instances
|
- several times, clients will be automaticaly balanced between instances
|
||||||
- with uvloop module to get an extra speed boost
|
- with uvloop module to get an extra speed boost
|
||||||
- with runtime statistics exported for [Prometheus](https://prometheus.io/): using [prometheus](https://github.com/alexbers/mtprotoproxy/tree/prometheus) branch
|
- with runtime statistics exported to [Prometheus](https://prometheus.io/)
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ SECURE_ONLY = True
|
|||||||
|
|
||||||
# Makes the proxy even more hard to detect
|
# Makes the proxy even more hard to detect
|
||||||
# Compatible only with the recent clients
|
# Compatible only with the recent clients
|
||||||
# TLS_ONLY = True
|
TLS_ONLY = True
|
||||||
|
|
||||||
# The domain for TLS, bad clients are proxied there
|
# The domain for TLS, bad clients are proxied there
|
||||||
# Use random existing domain, proxy checks it on start
|
# Use random existing domain, proxy checks it on start
|
||||||
|
|||||||
292
mtprotoproxy.py
292
mtprotoproxy.py
@@ -76,6 +76,7 @@ PADDING_FILLER = b"\x04\x00\x00\x00"
|
|||||||
MIN_MSG_LEN = 12
|
MIN_MSG_LEN = 12
|
||||||
MAX_MSG_LEN = 2 ** 24
|
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}
|
my_ip_info = {"ipv4": None, "ipv6": None}
|
||||||
used_handshakes = collections.OrderedDict()
|
used_handshakes = collections.OrderedDict()
|
||||||
@@ -83,6 +84,11 @@ disable_middle_proxy = False
|
|||||||
is_time_skewed = False
|
is_time_skewed = False
|
||||||
fake_cert_len = random.randrange(1024, 4096)
|
fake_cert_len = random.randrange(1024, 4096)
|
||||||
mask_host_cached_ip = None
|
mask_host_cached_ip = None
|
||||||
|
last_clients_with_time_skew = {}
|
||||||
|
last_clients_with_first_pkt_error = collections.Counter()
|
||||||
|
last_clients_with_same_handshake = collections.Counter()
|
||||||
|
proxy_start_time = 0
|
||||||
|
proxy_links = []
|
||||||
|
|
||||||
config = {}
|
config = {}
|
||||||
|
|
||||||
@@ -103,6 +109,10 @@ def init_config():
|
|||||||
conf_dict["USERS"] = {"user%d" % i: secrets[i].zfill(32) for i in range(len(secrets))}
|
conf_dict["USERS"] = {"user%d" % i: secrets[i].zfill(32) for i in range(len(secrets))}
|
||||||
if len(sys.argv) > 3:
|
if len(sys.argv) > 3:
|
||||||
conf_dict["AD_TAG"] = sys.argv[3]
|
conf_dict["AD_TAG"] = sys.argv[3]
|
||||||
|
if len(sys.argv) > 4:
|
||||||
|
conf_dict["TLS_DOMAIN"] = sys.argv[4]
|
||||||
|
conf_dict["TLS_ONLY"] = True
|
||||||
|
conf_dict["SECURE_ONLY"] = True
|
||||||
|
|
||||||
conf_dict = {k: v for k, v in conf_dict.items() if k.isupper()}
|
conf_dict = {k: v for k, v in conf_dict.items() if k.isupper()}
|
||||||
|
|
||||||
@@ -142,6 +152,16 @@ def init_config():
|
|||||||
# the next host's port to forward bad clients
|
# the next host's port to forward bad clients
|
||||||
conf_dict.setdefault("MASK_PORT", 443)
|
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
|
# user tcp connection limits, the mapping from name to the integer limit
|
||||||
# one client can create many tcp connections, up to 8
|
# one client can create many tcp connections, up to 8
|
||||||
conf_dict.setdefault("USER_MAX_TCP_CONNS", {})
|
conf_dict.setdefault("USER_MAX_TCP_CONNS", {})
|
||||||
@@ -156,10 +176,10 @@ def init_config():
|
|||||||
conf_dict.setdefault("USER_DATA_QUOTA", {})
|
conf_dict.setdefault("USER_DATA_QUOTA", {})
|
||||||
|
|
||||||
# length of used handshake randoms for active fingerprinting protection, zero to disable
|
# length of used handshake randoms for active fingerprinting protection, zero to disable
|
||||||
conf_dict.setdefault("REPLAY_CHECK_LEN", 32768)
|
conf_dict.setdefault("REPLAY_CHECK_LEN", 65536)
|
||||||
|
|
||||||
# block bad first packets to even more protect against replay-based fingerprinting
|
# block bad first packets to even more protect against replay-based fingerprinting
|
||||||
conf_dict.setdefault("BLOCK_IF_FIRST_PKT_BAD", True)
|
conf_dict.setdefault("BLOCK_IF_FIRST_PKT_BAD", not conf_dict["TLS_ONLY"])
|
||||||
|
|
||||||
# delay in seconds between stats printing
|
# delay in seconds between stats printing
|
||||||
conf_dict.setdefault("STATS_PRINT_PERIOD", 600)
|
conf_dict.setdefault("STATS_PRINT_PERIOD", 600)
|
||||||
@@ -201,10 +221,40 @@ def init_config():
|
|||||||
# listen unix socket
|
# listen unix socket
|
||||||
conf_dict.setdefault("LISTEN_UNIX_SOCK", "")
|
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)
|
||||||
|
|
||||||
# allow access to config by attributes
|
# allow access to config by attributes
|
||||||
config = type("config", (dict,), conf_dict)(conf_dict)
|
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():
|
def try_use_cryptography_module():
|
||||||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||||||
from cryptography.hazmat.backends import default_backend
|
from cryptography.hazmat.backends import default_backend
|
||||||
@@ -294,24 +344,45 @@ def print_err(*params):
|
|||||||
|
|
||||||
def init_stats():
|
def init_stats():
|
||||||
global stats
|
global stats
|
||||||
stats = {user: collections.Counter() for user in config.USERS}
|
global user_stats
|
||||||
|
|
||||||
|
stats = collections.Counter()
|
||||||
|
user_stats = {user: collections.Counter() for user in config.USERS}
|
||||||
|
|
||||||
|
|
||||||
def update_stats(user, connects=0, curr_connects=0, octets=0, msgs=0):
|
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
|
||||||
|
|
||||||
|
if user not in user_stats:
|
||||||
|
user_stats[user] = collections.Counter()
|
||||||
|
user_stats[user].update(**kw_stats)
|
||||||
|
|
||||||
|
|
||||||
|
def update_durations(duration):
|
||||||
global stats
|
global stats
|
||||||
|
|
||||||
if user not in stats:
|
for bucket in STAT_DURATION_BUCKETS:
|
||||||
stats[user] = collections.Counter()
|
if duration <= bucket:
|
||||||
|
break
|
||||||
|
|
||||||
stats[user].update(connects=connects, curr_connects=curr_connects,
|
update_stats(**{"connects_with_duration_le_%s" % str(bucket): 1})
|
||||||
octets=octets, msgs=msgs)
|
|
||||||
|
|
||||||
|
|
||||||
def get_curr_connects_count():
|
def get_curr_connects_count():
|
||||||
global stats
|
global user_stats
|
||||||
|
|
||||||
all_connects = 0
|
all_connects = 0
|
||||||
for user, stat in stats.items():
|
for user, stat in user_stats.items():
|
||||||
all_connects += stat["curr_connects"]
|
all_connects += stat["curr_connects"]
|
||||||
return all_connects
|
return all_connects
|
||||||
|
|
||||||
@@ -825,6 +896,8 @@ async def handle_bad_client(reader_clt, writer_clt, handshake):
|
|||||||
|
|
||||||
global mask_host_cached_ip
|
global mask_host_cached_ip
|
||||||
|
|
||||||
|
update_stats(connects_bad=1)
|
||||||
|
|
||||||
if writer_clt.transport.is_closing():
|
if writer_clt.transport.is_closing():
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -899,6 +972,8 @@ async def handle_bad_client(reader_clt, writer_clt, handshake):
|
|||||||
|
|
||||||
async def handle_fake_tls_handshake(handshake, reader, writer, peer):
|
async def handle_fake_tls_handshake(handshake, reader, writer, peer):
|
||||||
global used_handshakes
|
global used_handshakes
|
||||||
|
global last_clients_with_time_skew
|
||||||
|
global last_clients_with_same_handshake
|
||||||
global fake_cert_len
|
global fake_cert_len
|
||||||
|
|
||||||
TIME_SKEW_MIN = -20 * 60
|
TIME_SKEW_MIN = -20 * 60
|
||||||
@@ -910,6 +985,7 @@ async def handle_fake_tls_handshake(handshake, reader, writer, peer):
|
|||||||
TLS_APP_HTTP2_HDR = b"\x17" + TLS_VERS
|
TLS_APP_HTTP2_HDR = b"\x17" + TLS_VERS
|
||||||
|
|
||||||
DIGEST_LEN = 32
|
DIGEST_LEN = 32
|
||||||
|
DIGEST_HALFLEN = 16
|
||||||
DIGEST_POS = 11
|
DIGEST_POS = 11
|
||||||
|
|
||||||
SESSION_ID_LEN_POS = DIGEST_POS + DIGEST_LEN
|
SESSION_ID_LEN_POS = DIGEST_POS + DIGEST_LEN
|
||||||
@@ -920,8 +996,8 @@ async def handle_fake_tls_handshake(handshake, reader, writer, peer):
|
|||||||
|
|
||||||
digest = handshake[DIGEST_POS: DIGEST_POS + DIGEST_LEN]
|
digest = handshake[DIGEST_POS: DIGEST_POS + DIGEST_LEN]
|
||||||
|
|
||||||
if digest in used_handshakes:
|
if digest[:DIGEST_HALFLEN] in used_handshakes:
|
||||||
print_err("Active TLS fingerprinting detected from %s, handling it" % peer[0])
|
last_clients_with_same_handshake[peer[0]] += 1
|
||||||
return False
|
return False
|
||||||
|
|
||||||
sess_id_len = handshake[SESSION_ID_LEN_POS]
|
sess_id_len = handshake[SESSION_ID_LEN_POS]
|
||||||
@@ -944,8 +1020,7 @@ async def handle_fake_tls_handshake(handshake, reader, writer, peer):
|
|||||||
# some clients fail to read unix time and send the time since boot instead
|
# some clients fail to read unix time and send the time since boot instead
|
||||||
client_time_is_small = timestamp < 60*60*24*1000
|
client_time_is_small = timestamp < 60*60*24*1000
|
||||||
if not client_time_is_ok and not is_time_skewed and not client_time_is_small:
|
if not client_time_is_ok and not is_time_skewed and not client_time_is_small:
|
||||||
print_err("Client with time skew detected from %s, can be a replay-attack" % peer[0])
|
last_clients_with_time_skew[peer[0]] = (time.time() - timestamp) // 60
|
||||||
print_err("The clocks were %d minutes behind" % ((time.time() - timestamp) // 60))
|
|
||||||
continue
|
continue
|
||||||
|
|
||||||
http_data = myrandom.getrandbytes(fake_cert_len)
|
http_data = myrandom.getrandbytes(fake_cert_len)
|
||||||
@@ -967,7 +1042,7 @@ async def handle_fake_tls_handshake(handshake, reader, writer, peer):
|
|||||||
if config.REPLAY_CHECK_LEN > 0:
|
if config.REPLAY_CHECK_LEN > 0:
|
||||||
while len(used_handshakes) >= config.REPLAY_CHECK_LEN:
|
while len(used_handshakes) >= config.REPLAY_CHECK_LEN:
|
||||||
used_handshakes.popitem(last=False)
|
used_handshakes.popitem(last=False)
|
||||||
used_handshakes[digest] = True
|
used_handshakes[digest[:DIGEST_HALFLEN]] = True
|
||||||
|
|
||||||
reader = FakeTLSStreamReader(reader)
|
reader = FakeTLSStreamReader(reader)
|
||||||
writer = FakeTLSStreamWriter(writer)
|
writer = FakeTLSStreamWriter(writer)
|
||||||
@@ -997,7 +1072,7 @@ async def handle_proxy_protocol(reader, peer=None):
|
|||||||
if proxy_fam in (PROXY_TCP4, PROXY_TCP6):
|
if proxy_fam in (PROXY_TCP4, PROXY_TCP6):
|
||||||
if len(proxy_addr) == 4:
|
if len(proxy_addr) == 4:
|
||||||
src_addr = proxy_addr[0].decode('ascii')
|
src_addr = proxy_addr[0].decode('ascii')
|
||||||
src_port = proxy_addr[2].decode('ascii')
|
src_port = int(proxy_addr[2].decode('ascii'))
|
||||||
return (src_addr, src_port)
|
return (src_addr, src_port)
|
||||||
elif proxy_fam == PROXY_UNKNOWN:
|
elif proxy_fam == PROXY_UNKNOWN:
|
||||||
return peer
|
return peer
|
||||||
@@ -1033,6 +1108,7 @@ async def handle_proxy_protocol(reader, peer=None):
|
|||||||
|
|
||||||
async def handle_handshake(reader, writer):
|
async def handle_handshake(reader, writer):
|
||||||
global used_handshakes
|
global used_handshakes
|
||||||
|
global last_clients_with_same_handshake
|
||||||
|
|
||||||
TLS_START_BYTES = b"\x16\x03\x01\x02\x00\x01\x00\x01\xfc\x03\x03"
|
TLS_START_BYTES = b"\x16\x03\x01\x02\x00\x01\x00\x01\xfc\x03\x03"
|
||||||
|
|
||||||
@@ -1078,7 +1154,7 @@ async def handle_handshake(reader, writer):
|
|||||||
enc_prekey, enc_iv = enc_prekey_and_iv[:PREKEY_LEN], enc_prekey_and_iv[PREKEY_LEN:]
|
enc_prekey, enc_iv = enc_prekey_and_iv[:PREKEY_LEN], enc_prekey_and_iv[PREKEY_LEN:]
|
||||||
|
|
||||||
if dec_prekey_and_iv in used_handshakes:
|
if dec_prekey_and_iv in used_handshakes:
|
||||||
print_err("Active fingerprinting detected from %s, handling it" % peer[0])
|
last_clients_with_same_handshake[peer[0]] += 1
|
||||||
await handle_bad_client(reader, writer, handshake)
|
await handle_bad_client(reader, writer, handshake)
|
||||||
return False
|
return False
|
||||||
|
|
||||||
@@ -1364,10 +1440,13 @@ async def handle_client(reader_clt, writer_clt):
|
|||||||
set_ack_timeout(writer_clt.get_extra_info("socket"), config.CLIENT_ACK_TIMEOUT)
|
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())
|
set_bufsizes(writer_clt.get_extra_info("socket"), get_to_tg_bufsize(), get_to_clt_bufsize())
|
||||||
|
|
||||||
|
update_stats(connects_all=1)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
clt_data = await asyncio.wait_for(handle_handshake(reader_clt, writer_clt),
|
clt_data = await asyncio.wait_for(handle_handshake(reader_clt, writer_clt),
|
||||||
timeout=config.CLIENT_HANDSHAKE_TIMEOUT)
|
timeout=config.CLIENT_HANDSHAKE_TIMEOUT)
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
|
update_stats(handshake_timeouts=1)
|
||||||
return
|
return
|
||||||
|
|
||||||
if not clt_data:
|
if not clt_data:
|
||||||
@@ -1376,7 +1455,7 @@ async def handle_client(reader_clt, writer_clt):
|
|||||||
reader_clt, writer_clt, proto_tag, user, dc_idx, enc_key_and_iv, peer = clt_data
|
reader_clt, writer_clt, proto_tag, user, dc_idx, enc_key_and_iv, peer = clt_data
|
||||||
cl_ip, cl_port = peer
|
cl_ip, cl_port = peer
|
||||||
|
|
||||||
update_stats(user, connects=1)
|
update_user_stats(user, connects=1)
|
||||||
|
|
||||||
connect_directly = (not config.USE_MIDDLE_PROXY or disable_middle_proxy)
|
connect_directly = (not config.USE_MIDDLE_PROXY or disable_middle_proxy)
|
||||||
|
|
||||||
@@ -1419,6 +1498,7 @@ async def handle_client(reader_clt, writer_clt):
|
|||||||
return
|
return
|
||||||
|
|
||||||
async def connect_reader_to_writer(rd, wr, user, rd_buf_size, block_if_first_pkt_bad=False):
|
async def connect_reader_to_writer(rd, wr, user, rd_buf_size, block_if_first_pkt_bad=False):
|
||||||
|
global last_clients_with_first_pkt_error
|
||||||
is_first_pkt = True
|
is_first_pkt = True
|
||||||
try:
|
try:
|
||||||
while True:
|
while True:
|
||||||
@@ -1434,7 +1514,7 @@ async def handle_client(reader_clt, writer_clt):
|
|||||||
|
|
||||||
ERR_PKT_DATA = b'l\xfe\xff\xff'
|
ERR_PKT_DATA = b'l\xfe\xff\xff'
|
||||||
if block_if_first_pkt_bad and data == ERR_PKT_DATA:
|
if block_if_first_pkt_bad and data == ERR_PKT_DATA:
|
||||||
print_err("Active fingerprinting detected from %s, dropping it" % cl_ip)
|
last_clients_with_first_pkt_error[cl_ip] += 1
|
||||||
|
|
||||||
wr.write_eof()
|
wr.write_eof()
|
||||||
await wr.drain()
|
await wr.drain()
|
||||||
@@ -1445,7 +1525,7 @@ async def handle_client(reader_clt, writer_clt):
|
|||||||
await wr.drain()
|
await wr.drain()
|
||||||
return
|
return
|
||||||
else:
|
else:
|
||||||
update_stats(user, octets=len(data), msgs=1)
|
update_user_stats(user, octets=len(data), msgs=1)
|
||||||
wr.write(data, extra)
|
wr.write(data, extra)
|
||||||
await wr.drain()
|
await wr.drain()
|
||||||
except (OSError, asyncio.streams.IncompleteReadError) as e:
|
except (OSError, asyncio.streams.IncompleteReadError) as e:
|
||||||
@@ -1458,11 +1538,11 @@ async def handle_client(reader_clt, writer_clt):
|
|||||||
task_tg_to_clt = asyncio.ensure_future(tg_to_clt)
|
task_tg_to_clt = asyncio.ensure_future(tg_to_clt)
|
||||||
task_clt_to_tg = asyncio.ensure_future(clt_to_tg)
|
task_clt_to_tg = asyncio.ensure_future(clt_to_tg)
|
||||||
|
|
||||||
update_stats(user, curr_connects=1)
|
update_user_stats(user, curr_connects=1)
|
||||||
|
|
||||||
tcp_limit_hit = (
|
tcp_limit_hit = (
|
||||||
user in config.USER_MAX_TCP_CONNS and
|
user in config.USER_MAX_TCP_CONNS and
|
||||||
stats[user]["curr_connects"] > config.USER_MAX_TCP_CONNS[user]
|
user_stats[user]["curr_connects"] > config.USER_MAX_TCP_CONNS[user]
|
||||||
)
|
)
|
||||||
|
|
||||||
user_expired = (
|
user_expired = (
|
||||||
@@ -1472,13 +1552,15 @@ async def handle_client(reader_clt, writer_clt):
|
|||||||
|
|
||||||
user_data_quota_hit = (
|
user_data_quota_hit = (
|
||||||
user in config.USER_DATA_QUOTA and
|
user in config.USER_DATA_QUOTA and
|
||||||
stats[user]["octets"] > config.USER_DATA_QUOTA[user]
|
user_stats[user]["octets"] > config.USER_DATA_QUOTA[user]
|
||||||
)
|
)
|
||||||
|
|
||||||
if (not tcp_limit_hit) and (not user_expired) and (not user_data_quota_hit):
|
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)
|
await asyncio.wait([task_tg_to_clt, task_clt_to_tg], return_when=asyncio.FIRST_COMPLETED)
|
||||||
|
update_durations(time.time() - start)
|
||||||
|
|
||||||
update_stats(user, curr_connects=-1)
|
update_user_stats(user, curr_connects=-1)
|
||||||
|
|
||||||
task_tg_to_clt.cancel()
|
task_tg_to_clt.cancel()
|
||||||
task_clt_to_tg.cancel()
|
task_clt_to_tg.cancel()
|
||||||
@@ -1489,7 +1571,9 @@ async def handle_client(reader_clt, writer_clt):
|
|||||||
async def handle_client_wrapper(reader, writer):
|
async def handle_client_wrapper(reader, writer):
|
||||||
try:
|
try:
|
||||||
await handle_client(reader, writer)
|
await handle_client(reader, writer)
|
||||||
except (asyncio.IncompleteReadError, ConnectionResetError, TimeoutError):
|
except (asyncio.IncompleteReadError, asyncio.CancelledError):
|
||||||
|
pass
|
||||||
|
except (ConnectionResetError, TimeoutError):
|
||||||
pass
|
pass
|
||||||
except Exception:
|
except Exception:
|
||||||
traceback.print_exc()
|
traceback.print_exc()
|
||||||
@@ -1497,18 +1581,138 @@ async def handle_client_wrapper(reader, writer):
|
|||||||
writer.transport.abort()
|
writer.transport.abort()
|
||||||
|
|
||||||
|
|
||||||
async def stats_printer():
|
def make_metrics_pkt(metrics):
|
||||||
|
pkt_body_list = []
|
||||||
|
used_names = set()
|
||||||
|
|
||||||
|
for name, m_type, desc, val in metrics:
|
||||||
|
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("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 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_first_pkt_error
|
||||||
|
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"],
|
||||||
|
["user_msgs", "counter", "msgs proxied for user", "msgs"],
|
||||||
|
]
|
||||||
|
|
||||||
|
for m_name, m_type, m_desc, stat_key in user_metrics_desc:
|
||||||
|
for user, stat in user_stats.items():
|
||||||
|
metric = {"user": user, "val": stat[stat_key]}
|
||||||
|
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_clients_with_time_skew
|
||||||
|
global last_clients_with_first_pkt_error
|
||||||
|
global last_clients_with_same_handshake
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
await asyncio.sleep(config.STATS_PRINT_PERIOD)
|
await asyncio.sleep(config.STATS_PRINT_PERIOD)
|
||||||
|
|
||||||
print("Stats for", time.strftime("%d.%m.%Y %H:%M:%S"))
|
print("Stats for", time.strftime("%d.%m.%Y %H:%M:%S"))
|
||||||
for user, stat in stats.items():
|
for user, stat in user_stats.items():
|
||||||
print("%s: %d connects (%d current), %.2f MB, %d msgs" % (
|
print("%s: %d connects (%d current), %.2f MB, %d msgs" % (
|
||||||
user, stat["connects"], stat["curr_connects"],
|
user, stat["connects"], stat["curr_connects"],
|
||||||
stat["octets"] / 1000000, stat["msgs"]))
|
stat["octets"] / 1000000, stat["msgs"]))
|
||||||
print(flush=True)
|
print(flush=True)
|
||||||
|
|
||||||
|
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_first_pkt_error:
|
||||||
|
print("Clients with error on the first packet (possible replay-attackers):")
|
||||||
|
for ip, times in last_clients_with_first_pkt_error.items():
|
||||||
|
print("%s, %d times" % (ip, times))
|
||||||
|
print(flush=True)
|
||||||
|
last_clients_with_first_pkt_error.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"):
|
async def make_https_req(url, host="core.telegram.org"):
|
||||||
""" Make request, return resp body and headers. """
|
""" Make request, return resp body and headers. """
|
||||||
@@ -1601,6 +1805,8 @@ async def get_mask_host_cert_len():
|
|||||||
if cert:
|
if cert:
|
||||||
if len(cert) != fake_cert_len:
|
if len(cert) != fake_cert_len:
|
||||||
fake_cert_len = len(cert)
|
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:
|
else:
|
||||||
print_err("The MASK_HOST %s is not TLS 1.3 host, this is not recommended" %
|
print_err("The MASK_HOST %s is not TLS 1.3 host, this is not recommended" %
|
||||||
config.MASK_HOST)
|
config.MASK_HOST)
|
||||||
@@ -1753,6 +1959,7 @@ def init_ip_info():
|
|||||||
|
|
||||||
def print_tg_info():
|
def print_tg_info():
|
||||||
global my_ip_info
|
global my_ip_info
|
||||||
|
global proxy_links
|
||||||
|
|
||||||
print_default_warning = False
|
print_default_warning = False
|
||||||
|
|
||||||
@@ -1766,17 +1973,23 @@ def print_tg_info():
|
|||||||
if not ip_addrs:
|
if not ip_addrs:
|
||||||
ip_addrs = ["YOUR_IP"]
|
ip_addrs = ["YOUR_IP"]
|
||||||
|
|
||||||
|
proxy_links = []
|
||||||
|
|
||||||
for user, secret in sorted(config.USERS.items(), key=lambda x: x[0]):
|
for user, secret in sorted(config.USERS.items(), key=lambda x: x[0]):
|
||||||
for ip in ip_addrs:
|
for ip in ip_addrs:
|
||||||
if not config.TLS_ONLY:
|
if not config.TLS_ONLY:
|
||||||
if not config.SECURE_ONLY:
|
if not config.SECURE_ONLY:
|
||||||
params = {"server": ip, "port": config.PORT, "secret": secret}
|
params = {"server": ip, "port": config.PORT, "secret": secret}
|
||||||
params_encodeded = urllib.parse.urlencode(params, safe=':')
|
params_encodeded = urllib.parse.urlencode(params, safe=':')
|
||||||
print("{}: tg://proxy?{}".format(user, params_encodeded), flush=True)
|
classic_link = "tg://proxy?{}".format(params_encodeded)
|
||||||
|
proxy_links.append({"user": user, "link": classic_link})
|
||||||
|
print("{}: {}".format(user, classic_link), flush=True)
|
||||||
|
|
||||||
params = {"server": ip, "port": config.PORT, "secret": "dd" + secret}
|
params = {"server": ip, "port": config.PORT, "secret": "dd" + secret}
|
||||||
params_encodeded = urllib.parse.urlencode(params, safe=':')
|
params_encodeded = urllib.parse.urlencode(params, safe=':')
|
||||||
print("{}: tg://proxy?{}".format(user, params_encodeded), flush=True)
|
dd_link = "tg://proxy?{}".format(params_encodeded)
|
||||||
|
proxy_links.append({"user": user, "link": dd_link})
|
||||||
|
print("{}: {}".format(user, dd_link), flush=True)
|
||||||
|
|
||||||
tls_secret = "ee" + secret + config.TLS_DOMAIN.encode().hex()
|
tls_secret = "ee" + secret + config.TLS_DOMAIN.encode().hex()
|
||||||
# the base64 links is buggy on ios
|
# the base64 links is buggy on ios
|
||||||
@@ -1784,7 +1997,9 @@ def print_tg_info():
|
|||||||
# tls_secret_base64 = base64.urlsafe_b64encode(tls_secret)
|
# tls_secret_base64 = base64.urlsafe_b64encode(tls_secret)
|
||||||
params = {"server": ip, "port": config.PORT, "secret": tls_secret}
|
params = {"server": ip, "port": config.PORT, "secret": tls_secret}
|
||||||
params_encodeded = urllib.parse.urlencode(params, safe=':')
|
params_encodeded = urllib.parse.urlencode(params, safe=':')
|
||||||
print("{}: tg://proxy?{} (new)".format(user, params_encodeded), flush=True)
|
tls_link = "tg://proxy?{}".format(params_encodeded)
|
||||||
|
proxy_links.append({"user": user, "link": tls_link})
|
||||||
|
print("{}: {}".format(user, tls_link), flush=True)
|
||||||
|
|
||||||
if secret in ["00000000000000000000000000000000", "0123456789abcdef0123456789abcdef"]:
|
if secret in ["00000000000000000000000000000000", "0123456789abcdef0123456789abcdef"]:
|
||||||
msg = "The default secret {} is used, this is not recommended".format(secret)
|
msg = "The default secret {} is used, this is not recommended".format(secret)
|
||||||
@@ -1825,6 +2040,7 @@ def setup_signals():
|
|||||||
if hasattr(signal, 'SIGUSR2'):
|
if hasattr(signal, 'SIGUSR2'):
|
||||||
def reload_signal(signum, frame):
|
def reload_signal(signum, frame):
|
||||||
init_config()
|
init_config()
|
||||||
|
apply_upstream_proxy_settings()
|
||||||
print("Config reloaded", flush=True, file=sys.stderr)
|
print("Config reloaded", flush=True, file=sys.stderr)
|
||||||
print_tg_info()
|
print_tg_info()
|
||||||
|
|
||||||
@@ -1832,9 +2048,13 @@ def setup_signals():
|
|||||||
|
|
||||||
|
|
||||||
def try_setup_uvloop():
|
def try_setup_uvloop():
|
||||||
|
if config.SOCKS5_HOST and config.SOCKS5_PORT:
|
||||||
|
# socks mode is not compatible with uvloop
|
||||||
|
return
|
||||||
try:
|
try:
|
||||||
import uvloop
|
import uvloop
|
||||||
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
|
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
|
||||||
|
print_err("Found uvloop, using it for optimal performance")
|
||||||
except ImportError:
|
except ImportError:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@@ -1881,6 +2101,7 @@ def main():
|
|||||||
try_setup_uvloop()
|
try_setup_uvloop()
|
||||||
|
|
||||||
init_stats()
|
init_stats()
|
||||||
|
init_proxy_start_time()
|
||||||
|
|
||||||
if sys.platform == "win32":
|
if sys.platform == "win32":
|
||||||
loop = asyncio.ProactorEventLoop()
|
loop = asyncio.ProactorEventLoop()
|
||||||
@@ -1927,6 +2148,16 @@ def main():
|
|||||||
servers.append(loop.run_until_complete(task))
|
servers.append(loop.run_until_complete(task))
|
||||||
os.chmod(config.LISTEN_UNIX_SOCK, 0o666)
|
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, loop=loop)
|
||||||
|
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, loop=loop)
|
||||||
|
servers.append(loop.run_until_complete(task))
|
||||||
|
|
||||||
try:
|
try:
|
||||||
loop.run_forever()
|
loop.run_forever()
|
||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
@@ -1947,6 +2178,7 @@ def main():
|
|||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
init_config()
|
init_config()
|
||||||
|
apply_upstream_proxy_settings()
|
||||||
init_ip_info()
|
init_ip_info()
|
||||||
print_tg_info()
|
print_tg_info()
|
||||||
main()
|
main()
|
||||||
|
|||||||
Reference in New Issue
Block a user