https://pastein.ru/t/rDT
Загрузка данных
import os
import sys
import json
import base64
import binascii
import urllib.parse
import urllib.request
import subprocess
import winreg
import time
import ctypes
import atexit
import signal
import socket
import threading
from typing import Dict, List, Tuple, Optional, Any
from concurrent.futures import ThreadPoolExecutor
# --- НАСТРОЙКИ (Константы) ---
XRAY_EXE: str = "xray.exe"
KEY_FILE: str = "key.txt"
CONFIG_FILE: str = "config.json"
PROXY_HOST: str = "127.0.0.1"
HTTP_PORT: int = 10809
SOCKS_PORT: int = 10808
INTERNET_OPTION_SETTINGS_CHANGED: int = 39
INTERNET_OPTION_REFRESH: int = 37
HEALTH_CHECK_URLS: Tuple[str] = (
"https://cp.cloudflare.com/generate_204",
"https://www.gstatic.com/generate_204",
"https://clients3.google.com/generate_204"
)
# Глобальное состояние
global_process: Optional[subprocess.Popen] = None
process_lock: threading.Lock = threading.Lock()
stop_ping_monitor: threading.Event = threading.Event()
ping_thread: Optional[threading.Thread] = None
_cleanup_done: bool = False
_cleanup_lock: threading.Lock = threading.Lock()
# Бэкап настроек Windows Proxy
original_proxy_state: Dict[str, Any] = {
"enabled": 0,
"server": "",
"override": ""
}
proxy_backed_up: bool = False
_PROXY_HANDLER: urllib.request.ProxyHandler = urllib.request.ProxyHandler({
'http': f'http://{PROXY_HOST}:{HTTP_PORT}',
'https': f'http://{PROXY_HOST}:{HTTP_PORT}'
})
_PROXY_OPENER: urllib.request.OpenerDirector = urllib.request.build_opener(_PROXY_HANDLER)
def clear_console() -> None:
""" Очистка экрана консоли """
os.system('cls' if os.name == 'nt' else 'clear')
def refresh_wininet() -> None:
""" Уведомление Windows об изменении параметров сети через WinAPI """
try:
wininet = ctypes.windll.Wininet
wininet.InternetSetOptionW(0, INTERNET_OPTION_SETTINGS_CHANGED, 0, 0)
wininet.InternetSetOptionW(0, INTERNET_OPTION_REFRESH, 0, 0)
except (AttributeError, OSError) as e:
print(f"[!] Ошибка WinAPI: {e}")
def measure_ping(host: str, port: int, timeout: float = 5.0) -> Optional[int]:
""" Измерение задержки TCP-соединения (мс) """
try:
start_time: float = time.perf_counter()
with socket.create_connection((host, port), timeout=timeout):
return int((time.perf_counter() - start_time) * 1000)
except (socket.timeout, socket.error, OSError):
return None
def wait_for_port(host: str, port: int, timeout: float = 3.0) -> bool:
""" Проверка успешного открытия локального порта Xray """
start_time: float = time.perf_counter()
while time.perf_counter() - start_time < timeout:
try:
with socket.create_connection((host, port), timeout=0.2):
return True
except (socket.error, OSError):
time.sleep(0.1)
return False
def check_proxy_health(timeout: float = 4.0, retries: int = 2, delay: float = 1.0) -> bool:
""" Проверка прохождения HTTP-трафика через прокси по пулу сервисов """
for target_url in HEALTH_CHECK_URLS:
req = urllib.request.Request(target_url, headers={'User-Agent': 'Mozilla/5.0'})
for attempt in range(1, retries + 1):
try:
with _PROXY_OPENER.open(req, timeout=timeout) as response:
if response.status in (200, 204):
return True
except (urllib.error.URLError, urllib.error.HTTPError, socket.timeout, OSError):
pass
if attempt < retries:
time.sleep(delay)
return False
def load_vless_urls(file_path: str) -> List[str]:
""" Загрузка VLESS-ссылок (поддержка Base64 и обычной текстовой выгрузки) """
if not os.path.exists(file_path):
return []
try:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
content: str = f.read().strip()
except OSError:
return []
if not content:
return []
lines: List[str] = []
if not content.startswith("vless://"):
try:
padded_content: str = content + "=" * (-len(content) % 4)
decoded: str = base64.b64decode(padded_content).decode("utf-8", errors="ignore")
lines = decoded.splitlines()
except (binascii.Error, ValueError, UnicodeDecodeError):
lines = content.splitlines()
else:
lines = content.splitlines()
urls: List[str] = []
for line in lines:
cleaned: str = line.strip()
if cleaned.startswith("vless://"):
urls.append(cleaned)
return urls
def parse_vless_url(url_str: str, log_level: str = "warning") -> Tuple[Dict[str, Any], str, int]:
""" Парсинг VLESS ссылки с детальной валидацией параметров """
parsed = urllib.parse.urlparse(url_str)
if parsed.scheme != "vless":
raise ValueError("Ссылка должна начинаться с vless://")
uuid: str = parsed.username or ""
if not uuid:
raise ValueError("UUID пользователя отсутствует в VLESS-ссылке!")
target_host: str = parsed.hostname or ""
if not target_host:
raise ValueError("В VLESS-ссылке не указан целевой хост сервера!")
port: int = parsed.port or 443
params: Dict[str, str] = dict(urllib.parse.parse_qsl(parsed.query))
stream_type: str = params.get("type", "tcp").lower()
security: str = params.get("security", "none").lower()
sni: str = params.get("sni", target_host)
host_header: str = params.get("host", "")
fp: str = params.get("fp", "chrome")
pbk: str = params.get("pbk", "")
sid: str = params.get("sid", "")
flow: str = params.get("flow", "")
path: str = urllib.parse.unquote(params.get("path", "/"))
service_name: str = params.get("serviceName", "")
alpn_raw: str = params.get("alpn", "")
spider_x: str = params.get("spiderx", "/")
header_type: str = params.get("headerType", "none")
encryption: str = params.get("encryption", "none")
seed: str = params.get("seed", "")
quic_security: str = params.get("quicSecurity", "none")
key: str = params.get("key", "")
mode: str = params.get("mode", "gun")
stream_settings: Dict[str, Any] = {
"network": stream_type,
"security": security
}
# Безопасность
if security == "reality":
if not pbk:
raise ValueError("Для протокола REALITY обязателен параметр publicKey (pbk)!")
reality_settings: Dict[str, Any] = {
"serverName": sni,
"fingerprint": fp if fp else "chrome",
"publicKey": pbk,
"spiderX": spider_x
}
if sid:
reality_settings["shortId"] = sid
stream_settings["realitySettings"] = reality_settings
elif security == "tls":
tls_settings: Dict[str, Any] = {
"serverName": sni,
"allowInsecure": False,
"fingerprint": fp if fp else "chrome"
}
if alpn_raw:
tls_settings["alpn"] = [a.strip() for a in alpn_raw.split(",")]
stream_settings["tlsSettings"] = tls_settings
# Транспорт
if stream_type == "grpc":
stream_settings["grpcSettings"] = {
"serviceName": service_name,
"multiMode": (mode == "multi")
}
elif stream_type in ("ws", "xhttp"):
ws_settings: Dict[str, Any] = {"path": path}
if host_header:
ws_settings["headers"] = {"Host": host_header}
stream_settings[f"{stream_type}Settings"] = ws_settings
elif stream_type in ("h2", "http"):
stream_settings["network"] = "h2"
h2_settings: Dict[str, Any] = {"path": path}
if host_header:
h2_settings["host"] = [h.strip() for h in host_header.split(",")]
stream_settings["h2Settings"] = h2_settings
elif stream_type == "kcp":
kcp_settings: Dict[str, Any] = {
"header": {"type": header_type}
}
if seed:
kcp_settings["seed"] = seed
stream_settings["kcpSettings"] = kcp_settings
elif stream_type == "quic":
stream_settings["quicSettings"] = {
"security": quic_security,
"key": key,
"header": {"type": header_type}
}
elif stream_type == "tcp" and header_type == "http":
tcp_settings: Dict[str, Any] = {
"header": {
"type": "http",
"request": {"path": [path]}
}
}
if host_header:
tcp_settings["header"]["request"]["headers"] = {"Host": [host_header]}
stream_settings["tcpSettings"] = tcp_settings
user_data: Dict[str, str] = {"id": uuid, "encryption": encryption}
if flow:
user_data["flow"] = flow
config: Dict[str, Any] = {
"log": {"loglevel": log_level},
"inbounds": [
{"port": HTTP_PORT, "listen": PROXY_HOST, "protocol": "http", "settings": {"timeout": 300}},
{"port": SOCKS_PORT, "listen": PROXY_HOST, "protocol": "socks", "settings": {"udp": True}}
],
"outbounds": [
{
"protocol": "vless",
"settings": {
"vnext": [
{
"address": target_host,
"port": port,
"users": [user_data]
}
]
},
"streamSettings": stream_settings
}
]
}
return config, target_host, port
def backup_windows_proxy() -> None:
""" Бэкап прокси системных настроек Windows """
global proxy_backed_up, original_proxy_state
reg_path: str = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings"
try:
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, reg_path, 0, winreg.KEY_READ) as registry:
try:
original_proxy_state["enabled"] = winreg.QueryValueEx(registry, "ProxyEnable")[0]
except FileNotFoundError:
original_proxy_state["enabled"] = 0
try:
original_proxy_state["server"] = winreg.QueryValueEx(registry, "ProxyServer")[0]
except FileNotFoundError:
original_proxy_state["server"] = ""
try:
original_proxy_state["override"] = winreg.QueryValueEx(registry, "ProxyOverride")[0]
except FileNotFoundError:
original_proxy_state["override"] = ""
proxy_backed_up = True
except OSError as e:
print(f"[!] Ошибка сохранения исходных настроек прокси: {e}")
def set_windows_proxy(enable: bool = True) -> None:
""" Включение / Восстановление прокси Windows """
global proxy_backed_up
reg_path: str = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings"
try:
if enable and not proxy_backed_up:
backup_windows_proxy()
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, reg_path, 0, winreg.KEY_WRITE) as registry:
if enable:
winreg.SetValueEx(registry, "ProxyEnable", 0, winreg.REG_DWORD, 1)
winreg.SetValueEx(registry, "ProxyServer", 0, winreg.REG_SZ, f"{PROXY_HOST}:{HTTP_PORT}")
winreg.SetValueEx(registry, "ProxyOverride", 0, winreg.REG_SZ, "<local>;127.*;10.*;172.16.*;192.168.*")
print(f"\n[+] Системный прокси Windows: ВКЛЮЧЕН ({PROXY_HOST}:{HTTP_PORT})")
else:
if proxy_backed_up:
winreg.SetValueEx(registry, "ProxyEnable", 0, winreg.REG_DWORD, original_proxy_state["enabled"])
winreg.SetValueEx(registry, "ProxyServer", 0, winreg.REG_SZ, original_proxy_state["server"])
winreg.SetValueEx(registry, "ProxyOverride", 0, winreg.REG_SZ, original_proxy_state["override"])
print("\n[*] Системный прокси Windows: ВОССТАНОВЛЕН в исходное состояние.")
else:
winreg.SetValueEx(registry, "ProxyEnable", 0, winreg.REG_DWORD, 0)
winreg.SetValueEx(registry, "ProxyServer", 0, winreg.REG_SZ, "")
refresh_wininet()
except PermissionError:
print("[!] Ошибка доступа к реестру Windows.")
except OSError as e:
print(f"[!] Ошибка реестра Windows: {e}")
def cleanup() -> None:
""" Идемпотентное и безопасное завершение работы """
global _cleanup_done, global_process, ping_thread
with _cleanup_lock:
if _cleanup_done:
return
_cleanup_done = True
stop_ping_monitor.set()
if ping_thread and ping_thread.is_alive():
ping_thread.join(timeout=1.0)
set_windows_proxy(False)
with process_lock:
if global_process and global_process.poll() is None:
global_process.terminate()
try:
global_process.wait(timeout=2)
except subprocess.TimeoutExpired:
global_process.kill()
global_process = None
if os.path.exists(CONFIG_FILE):
for _ in range(5):
try:
os.remove(CONFIG_FILE)
break
except OSError:
time.sleep(0.1)
def _win_ctrl_handler(ctrl_type: int) -> bool:
cleanup()
os._exit(0)
def setup_os_handlers() -> None:
atexit.register(cleanup)
signal.signal(signal.SIGINT, lambda s, f: sys.exit(0))
signal.signal(signal.SIGTERM, lambda s, f: sys.exit(0))
if sys.platform == "win32":
PHANDLER_ROUTINE = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_ulong)
global _handler_ref
_handler_ref = PHANDLER_ROUTINE(_win_ctrl_handler)
ctypes.windll.kernel32.SetConsoleCtrlHandler(_handler_ref, True)
def get_server_metadata(url_str: str) -> Optional[Dict[str, Any]]:
try:
parsed = urllib.parse.urlparse(url_str)
host: Optional[str] = parsed.hostname
port: int = parsed.port or 443
if not host:
return None
name: str = urllib.parse.unquote(parsed.fragment) if parsed.fragment else f"{host}:{port}"
ping: Optional[int] = measure_ping(host, port)
return {
"name": name,
"host": host,
"port": port,
"ping": ping,
"url": url_str
}
except Exception:
return None
def select_server(vless_urls: List[str]) -> Optional[str]:
while True:
print("\n[*] Измерение задержки серверов...")
max_workers = max(1, min(len(vless_urls), 20))
with ThreadPoolExecutor(max_workers=max_workers) as executor:
results = executor.map(get_server_metadata, vless_urls)
servers = [res for res in results if res is not None]
if not servers:
print("[!] Нет доступных серверов.")
return None
servers.sort(key=lambda x: (x['ping'] is None, x['ping']))
print("\nСписок серверов:")
print("-" * 60)
for idx, srv in enumerate(servers, start=1):
ping_str: str = f"{srv['ping']} ms" if srv['ping'] is not None else "TIMEOUT"
print(f" [{idx}] {srv['name']:<30} | Пинг: {ping_str}")
print(" [R] Обновить пинг")
print("-" * 60)
choice: str = input(f"Выберите номер сервера (1-{len(servers)}) или 'R' для обновления: ").strip()
if choice.lower() in ("r", "refresh", "к"):
continue
try:
idx: int = int(choice) - 1
if 0 <= idx < len(servers):
return str(servers[idx]['url'])
print("[!] Неверный номер.")
except ValueError:
print("[!] Неверный ввод.")
except (KeyboardInterrupt, EOFError):
return None
def live_ping_monitor(host: str, port: int, show_logs: bool) -> None:
""" Динамический мониторинг задержки """
while not stop_ping_monitor.is_set():
ping: Optional[int] = measure_ping(host, port)
ping_str: str = f"{ping} ms" if ping is not None else "TIMEOUT"
if not show_logs:
msg: str = f"\r[STATUS] {host}:{port} | Пинг: {ping_str:<8} | Enter — выход"
sys.stdout.write(msg.ljust(75))
sys.stdout.flush()
stop_ping_monitor.wait(3.0)
if not show_logs:
sys.stdout.write("\r" + " " * 75 + "\r")
sys.stdout.flush()
def main() -> None:
global global_process, ping_thread
setup_os_handlers()
print("=" * 60)
print(" МИКРО VPN-КЛИЕНТ (XRAY CORE)")
print("=" * 60)
if not os.path.exists(XRAY_EXE):
print(f"\n[!] ОШИБКА: Файл '{XRAY_EXE}' не найден!")
input("\nНажмите Enter для выхода...")
return
if not os.path.exists(KEY_FILE):
print(f"\n[!] ОШИБКА: Файл '{KEY_FILE}' не найден!")
input("\nНажмите Enter для выхода...")
return
vless_urls: List[str] = load_vless_urls(KEY_FILE)
if not vless_urls:
print(f"\n[!] В '{KEY_FILE}' не найдено VLESS-ссылок!")
input("\nНажмите Enter для выхода...")
return
selected_url: Optional[str] = select_server(vless_urls)
if not selected_url:
return
show_logs: bool = input("\nПоказывать логи Xray в консоли? (y/n): ").strip().lower() in ("y", "yes", "д", "да")
log_level: str = "info" if show_logs else "warning"
clear_console()
try:
config_data, target_host, target_port = parse_vless_url(selected_url, log_level=log_level)
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(config_data, f, indent=2)
print("[+] Конфигурация сформирована.")
except Exception as e:
print(f"\n[!] Ошибка парсинга ссылки: {e}")
input("\nНажмите Enter для выхода...")
return
try:
print("[*] Запуск Xray-core...")
stdout_dest = None if show_logs else subprocess.DEVNULL
stderr_dest = None if show_logs else subprocess.DEVNULL
with process_lock:
global_process = subprocess.Popen(
[XRAY_EXE, "run", "-c", CONFIG_FILE],
stdout=stdout_dest,
stderr=stderr_dest
)
# Ожидание открытия сокета входящего прокси-порта
if not wait_for_port(PROXY_HOST, HTTP_PORT, timeout=3.0):
print("\n[!] ОШИБКА: Xray-core не смог открыть порт прокси.")
print("[!] Проверьте занятость портов 10808 / 10809.")
cleanup()
input("\nНажмите Enter для выхода...")
return
if global_process.poll() is not None:
print("\n[!] ОШИБКА: Xray-core завершил работу сразу после запуска.")
cleanup()
input("\nНажмите Enter для выхода...")
return
print("[*] Проверка проходимости трафика через прокси...")
if not check_proxy_health(timeout=4.0, retries=2):
print("\n[!] ОШИБКА: Трафик не проходит через прокси.")
print("[!] Ключ недействителен или соединение блокируется.")
cleanup()
input("\nНажмите Enter для выхода...")
return
set_windows_proxy(True)
stop_ping_monitor.clear()
ping_thread = threading.Thread(
target=live_ping_monitor,
args=(target_host, target_port, show_logs),
daemon=True
)
ping_thread.start()
if show_logs:
print("\nНажмите Enter для отключения...")
input()
except KeyboardInterrupt:
pass
except Exception as e:
print(f"[!] Ошибка исполнения: {e}")
finally:
print("\n[*] Завершение работы и сброс прокси...")
cleanup()
print("[+] Готово.")
if __name__ == "__main__":
main()