#!/usr/bin/env python3 import sys if sys.version_info < (3, 8): sys.stderr.write( "Ошибка: proxy-client требует Python 3.8 или новее; обнаружен Python {}.{}.\n".format( sys.version_info[0], sys.version_info[1] ) ) sys.exit(1) import argparse import configparser import fcntl import hashlib import http.client import ipaddress import json import os import platform import re import shutil import signal import socket import stat import subprocess import tempfile import time import urllib.error import urllib.request import uuid import zipfile from pathlib import Path from urllib.parse import parse_qsl, unquote, urlsplit CONFIG_KEYS = {"type", "name", "address", "port", "no_proxy_extra"} DEFAULT_NAME = "Hiddify" DEFAULT_ADDRESS = "hproxy.local" DEFAULT_PORT = 12334 DEFAULT_NO_PROXY_EXTRA = (".concept-software.ru",) XRAY_VERSION = "26.3.27" VERSION_LINE = re.compile( r"^Xray ({0})(?: .*)?$".format(re.escape(XRAY_VERSION)) ) XRAY_RELEASE_BASE_URL = ( "https://github.com/XTLS/Xray-core/releases/download/v{}/".format(XRAY_VERSION) ) MAX_ARCHIVE_BYTES = 64 * 1024 * 1024 MAX_DIGEST_BYTES = 4 * 1024 MAX_XRAY_BYTES = 64 * 1024 * 1024 STREAM_CHUNK_BYTES = 64 * 1024 QUERY_KEYS = { "type", "security", "encryption", "flow", "sni", "fp", "pbk", "sid", "headerType", "host", "alpn", "hiddify", "core", "path", "mode", "extra", "pcs", "vcn", "allowInsecure", "insecure", } ALPN_VALUES = {"http/1.1", "h2", "h3"} STATE_KEYS = { "supervisor_pid", "xray_pid", "xray_path", "config_path", "address", "port", "supervisor_path", "supervisor_starttime", "xray_starttime", "boot_id", } class ProxyError(Exception): pass class StartLockBusyError(ProxyError): pass class ProxyConfig: def __init__(self, proxy_type, name, address, port, no_proxy_extra): self.proxy_type = proxy_type self.name = name self.address = address self.port = port self.no_proxy_extra = no_proxy_extra class VlessProfile: def __init__(self, server, port, profile_id, query, alpn, extra): self.server = server self.port = port self.profile_id = profile_id self.query = query self.alpn = alpn self.extra = extra @property def transport(self): return self.query["type"] @property def security(self): return self.query["security"] def config_path(): config_home = os.environ.get("XDG_CONFIG_HOME") if config_home: return Path(config_home).expanduser() / "proxy-client/config" return Path.home() / ".config/proxy-client/config" def proxy_client_config_home(): config_home = os.environ.get("XDG_CONFIG_HOME") if config_home: return Path(config_home).expanduser() / "proxy-client" return Path.home() / ".config/proxy-client" def state_home(): state_root = os.environ.get("XDG_STATE_HOME") if state_root: return Path(state_root).expanduser() / "proxy-client" return Path.home() / ".local/state/proxy-client" def state_path(): return state_home() / "state.json" def generated_config_path(): return state_home() / "client.json" def lock_path(): return state_home() / "lock" def validate_address(value): try: parsed = ipaddress.ip_address(value) except ValueError: pass else: if "%" in value: raise ProxyError("поле address не поддерживает IPv6 zone ID") return parsed if ( not value or not value.isascii() or any(character.isspace() for character in value) or any(separator in value for separator in ("://", "@", "/", "[", "]", "%")) ): raise ProxyError("поле address должно быть IP-адресом или простым ASCII-именем") return None def parse_no_proxy_extra(value): if value == "": return () items = tuple(item.strip() for item in value.split(",")) if any(not item for item in items): raise ProxyError("no_proxy_extra содержит пустой элемент") return items def default_config(): return ProxyConfig( "external", DEFAULT_NAME, DEFAULT_ADDRESS, DEFAULT_PORT, DEFAULT_NO_PROXY_EXTRA, ) def load_config(): path = config_path() if not path.exists(): if path.is_symlink(): raise ProxyError("машинный конфиг ссылается на отсутствующий файл: {}".format(path)) return default_config() if not path.is_file(): raise ProxyError("машинный конфиг не является обычным файлом: {}".format(path)) try: text = path.read_text(encoding="utf-8") except OSError as error: raise ProxyError("не удалось прочитать машинный конфиг {}: {}".format(path, error)) parser = configparser.ConfigParser( interpolation=None, strict=True, allow_no_value=False, empty_lines_in_values=False, ) try: parser.read_string(text, source=str(path)) except configparser.Error as error: raise ProxyError("неверный машинный INI: {}".format(error)) if parser.sections() != ["proxy"]: raise ProxyError("машинный INI должен содержать только секцию [proxy]") section = parser["proxy"] actual_keys = set(section.keys()) if actual_keys != CONFIG_KEYS: missing = sorted(CONFIG_KEYS - actual_keys) unknown = sorted(actual_keys - CONFIG_KEYS) details = [] if missing: details.append("нет ключей: {}".format(", ".join(missing))) if unknown: details.append("неизвестные ключи: {}".format(", ".join(unknown))) raise ProxyError("неверные ключи [proxy] ({})".format("; ".join(details))) values = {key: section.get(key, raw=True) for key in CONFIG_KEYS} proxy_type = values["type"].strip() if proxy_type not in ("external", "xray"): raise ProxyError("поле type должно быть external или xray") name = values["name"].strip() address = values["address"].strip() if not name: raise ProxyError("поле name не должно быть пустым") if not address: raise ProxyError("поле address не должно быть пустым") validate_address(address) port_text = values["port"].strip() if not re.fullmatch(r"[0-9]+", port_text): raise ProxyError("поле port должно быть целым числом от 1 до 65535") port = int(port_text) if not 1 <= port <= 65535: raise ProxyError("поле port должно быть в диапазоне 1–65535") no_proxy_extra = parse_no_proxy_extra(values["no_proxy_extra"]) return ProxyConfig(proxy_type, name, address, port, no_proxy_extra) def preflight_timeout(): raw = os.environ.get("PROXY_PREFLIGHT_TIMEOUT", "2") try: timeout = float(raw) if not timeout > 0: raise ValueError except ValueError: raise ProxyError("PROXY_PREFLIGHT_TIMEOUT должен быть положительным числом секунд") return timeout def check_reachable(config, timeout): connection = None try: connection = socket.create_connection((config.address, config.port), timeout=timeout) except OSError as error: raise ProxyError( "прокси {} ({}:{}) недоступен: {}".format( config.name, config.address, config.port, error ) ) finally: if connection is not None: connection.close() def check_endpoint(address, port, name="управляемый Xray"): config = ProxyConfig("xray", name, address, port, ()) check_reachable(config, preflight_timeout()) def proxy_host_for_url(address): try: parsed = ipaddress.ip_address(address) except ValueError: return address if parsed.version == 6: return "[{}]".format(address) return address def environment_values(config): host = proxy_host_for_url(config.address) http_url = "http://{}:{}".format(host, config.port) socks_url = "socks5://{}:{}".format(host, config.port) no_proxy_items = [config.address, "localhost", "127.0.0.1", "::1", "[::1]"] no_proxy_items.extend(config.no_proxy_extra) no_proxy = ",".join(dict.fromkeys(no_proxy_items)) return [ ("HTTP_PROXY", http_url), ("HTTPS_PROXY", http_url), ("ALL_PROXY", socks_url), ("NO_PROXY", no_proxy), ("http_proxy", http_url), ("https_proxy", http_url), ("all_proxy", socks_url), ("no_proxy", no_proxy), ] def private_directory(path): path.mkdir(parents=True, exist_ok=True, mode=0o700) try: path.chmod(0o700) except OSError as error: raise ProxyError("не удалось закрыть служебный каталог: {}".format(error)) def write_private_json(path, value): private_directory(path.parent) try: descriptor = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as stream: json.dump( value, stream, ensure_ascii=False, sort_keys=True, allow_nan=False, ) stream.write("\n") path.chmod(0o600) except (OSError, ValueError) as error: raise ProxyError("не удалось записать закрытый служебный файл: {}".format(error)) def read_profile(path): try: resolved = path.expanduser().resolve(strict=True) file_stat = resolved.stat() except (OSError, RuntimeError) as error: raise ProxyError("не удалось открыть файл профиля: {}".format(error)) if not stat.S_ISREG(file_stat.st_mode): raise ProxyError("файл профиля должен быть обычным файлом") if file_stat.st_mode & 0o077: raise ProxyError( "конечный файл профиля доступен группе или другим пользователям; выполните chmod 600" ) try: text = resolved.read_text(encoding="utf-8") except OSError as error: raise ProxyError("не удалось прочитать файл профиля: {}".format(error)) lines = [line.strip() for line in text.splitlines() if line.strip()] if len(lines) != 1: raise ProxyError("файл профиля должен содержать ровно одну непустую VLESS-ссылку") return lines[0] def parse_query(raw_query): try: pairs = parse_qsl( raw_query, keep_blank_values=True, strict_parsing=True, encoding="utf-8", errors="strict", ) except (UnicodeError, ValueError): raise ProxyError("профиль содержит неверную строку параметров") query = {} for key, value in pairs: if key in query: raise ProxyError("профиль содержит повтор query-параметра") if key not in QUERY_KEYS: raise ProxyError("профиль содержит неизвестный query-параметр") query[key] = value return query def required(query, *keys): missing = [key for key in keys if not query.get(key)] if missing: raise ProxyError("в профиле отсутствуют обязательные параметры") def parse_boolean_fields(query): values = [] for key in ("allowInsecure", "insecure"): if key in query: if query[key] not in ("true", "false"): raise ProxyError("legacy-поля принимают только true или false") values.append(query[key] == "true") if len(values) == 2 and values[0] != values[1]: raise ProxyError("allowInsecure и insecure должны совпадать") return values[0] if values else False def parse_json_object(value): def reject_duplicates(pairs): result = {} for key, item in pairs: if key in result: raise ProxyError("extra содержит повтор ключа JSON") result[key] = item return result def reject_constant(_constant): raise ProxyError("extra содержит константу вне стандарта JSON") try: parsed = json.loads( value, object_pairs_hook=reject_duplicates, parse_constant=reject_constant, ) except ProxyError: raise except (TypeError, ValueError): raise ProxyError("extra должен быть корректным JSON-объектом") if not isinstance(parsed, dict): raise ProxyError("extra должен быть JSON-объектом") return parsed def parse_vless(url): try: parsed = urlsplit(url) except ValueError: raise ProxyError("неверная VLESS-ссылка") if ( parsed.scheme != "vless" or not parsed.netloc or parsed.password is not None or parsed.path ): raise ProxyError("ожидалась одна ссылка vless://UUID@SERVER:PORT") try: profile_id = unquote(parsed.username or "", encoding="utf-8", errors="strict") server = parsed.hostname port = parsed.port except (UnicodeError, ValueError): raise ProxyError("неверные UUID, сервер или порт VLESS") try: parsed_uuid = uuid.UUID(profile_id) except ValueError: raise ProxyError("профиль содержит неверный UUID") if str(parsed_uuid) != profile_id.lower() or not server or port is None: raise ProxyError("профиль содержит неверные UUID, сервер или порт") validate_address(server) query = parse_query(parsed.query) required(query, "type", "security", "encryption", "sni", "fp") if query["encryption"] != "none": raise ProxyError("поддерживается только encryption=none") if query.get("flow") not in (None, "xtls-rprx-vision"): raise ProxyError("поддерживается только flow=xtls-rprx-vision") if query.get("headerType", "none") != "none": raise ProxyError("поддерживается только headerType=none") if "hiddify" in query and query["hiddify"] != "1": raise ProxyError("поддерживается только метаданные hiddify=1") if "core" in query and query["core"] != "xray": raise ProxyError("поддерживается только метаданные core=xray") legacy_insecure = parse_boolean_fields(query) alpn = None extra = None if query["type"] == "tcp" and query["security"] == "reality": required(query, "flow", "pbk", "sid") forbidden = {"path", "mode", "extra", "pcs", "vcn", "core", "allowInsecure", "insecure"} if forbidden.intersection(query): raise ProxyError("параметры профиля несовместимы с TCP/REALITY") elif query["type"] == "xhttp" and query["security"] == "tls": required(query, "host", "path") if "flow" in query or "pbk" in query or "sid" in query: raise ProxyError("параметры профиля несовместимы с XHTTP/TLS") if "pcs" in query: hashes = list_value(query["pcs"]) if any(re.fullmatch(r"[0-9A-Fa-f]{64}", item) is None for item in hashes): raise ProxyError("pcs должен содержать 32-байтные SHA-256 в hex") if "vcn" in query: list_value(query["vcn"]) if "alpn" in query: alpn = list_value(query["alpn"]) if not alpn or any(item not in ALPN_VALUES for item in alpn): raise ProxyError("профиль содержит неподдерживаемый ALPN") if "extra" in query: extra = parse_json_object(query["extra"]) else: raise ProxyError("поддерживаются только TCP/REALITY и XHTTP/TLS") profile = VlessProfile(server, port, profile_id, query, alpn, extra) profile.legacy_insecure = legacy_insecure return profile def list_value(value): items = value.split(",") if any(not item for item in items): raise ProxyError("списочный параметр содержит пустой элемент") return items def build_xray_config(profile, local_address, local_port): query = profile.query settings = { "address": profile.server, "port": profile.port, "id": profile.profile_id, "encryption": query["encryption"], } if "flow" in query: settings["flow"] = query["flow"] if profile.transport == "tcp": reality = { "serverName": query["sni"], "fingerprint": query["fp"], "password": query["pbk"], "shortId": query["sid"], } stream = { "network": "raw", "security": "reality", "realitySettings": reality, } else: xhttp = {"host": query["host"], "path": query["path"]} if "mode" in query: xhttp["mode"] = query["mode"] if profile.extra is not None: xhttp["extra"] = profile.extra tls = {"serverName": query["sni"], "fingerprint": query["fp"]} if profile.alpn is not None: tls["alpn"] = profile.alpn if "pcs" in query: tls["pinnedPeerCertSha256"] = query["pcs"] if "vcn" in query: tls["verifyPeerCertByName"] = query["vcn"] stream = { "network": "xhttp", "security": "tls", "xhttpSettings": xhttp, "tlsSettings": tls, } return { "log": {"access": "none", "loglevel": "warning"}, "inbounds": [ { "tag": "local-proxy", "listen": local_address, "port": local_port, "protocol": "socks", "settings": {"auth": "noauth", "udp": True}, } ], "outbounds": [ { "tag": "proxy", "protocol": "vless", "settings": settings, "streamSettings": stream, } ], } def resolve_xray(): found = shutil.which("xray") if found is None: raise ProxyError("xray не найден в PATH; сначала выполните proxy-client install") try: return str(Path(found).resolve(strict=True)) except OSError as error: raise ProxyError("не удалось разрешить путь к xray: {}".format(error)) def xray_version(path): try: completed = subprocess.run( [path, "version"], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=5, check=False, ) except (OSError, subprocess.TimeoutExpired): raise ProxyError("не удалось безопасно определить версию выбранного xray") first_line = completed.stdout.splitlines()[0] if completed.stdout.splitlines() else "" match = VERSION_LINE.fullmatch(first_line) if completed.returncode != 0 or match is None: raise ProxyError("выбранный xray не сообщил точно поддерживаемую версию") return match.group(1) def xray_asset_for_architecture(machine): normalized = machine.lower() if normalized in ("x86_64", "amd64"): return "Xray-linux-64.zip" if normalized in ("aarch64", "arm64"): return "Xray-linux-arm64-v8a.zip" raise ProxyError("архитектура {} не поддерживается установщиком Xray".format(machine)) def download_file(url, destination, max_bytes): try: with urllib.request.urlopen(url, timeout=30) as response: raw_length = response.headers.get("Content-Length") if raw_length is None: declared_length = None elif re.fullmatch(r"[0-9]+", raw_length) is None: raise ProxyError("сервер сообщил неверный Content-Length") else: declared_length = int(raw_length) if declared_length > max_bytes: raise ProxyError("скачиваемый файл превышает допустимый размер") with destination.open("wb") as stream: actual_length = 0 while True: read_size = min( STREAM_CHUNK_BYTES, max_bytes - actual_length + 1 ) chunk = response.read(read_size) if not chunk: break actual_length += len(chunk) if actual_length > max_bytes: raise ProxyError("скачиваемый файл превышает допустимый размер") if declared_length is not None and actual_length > declared_length: raise ProxyError( "фактический размер не совпадает с Content-Length" ) stream.write(chunk) if declared_length is not None and actual_length != declared_length: raise ProxyError("фактический размер не совпадает с Content-Length") except ProxyError: raise except (OSError, urllib.error.URLError, http.client.HTTPException) as error: raise ProxyError("не удалось скачать официальный Xray: {}".format(error)) def expected_archive_sha256(digest_path): try: with digest_path.open("rb") as stream: digest_bytes = stream.read(MAX_DIGEST_BYTES + 1) except (OSError, UnicodeError) as error: raise ProxyError("не удалось прочитать файл SHA-256: {}".format(error)) if len(digest_bytes) > MAX_DIGEST_BYTES: raise ProxyError("файл SHA-256 превышает допустимый размер") try: digest_text = digest_bytes.decode("ascii") except UnicodeError as error: raise ProxyError("не удалось прочитать файл SHA-256: {}".format(error)) lines = digest_text.split("\n") if lines and lines[-1] == "": lines.pop() expected_lengths = { "MD5": 32, "SHA1": 40, "SHA2-256": 64, "SHA2-512": 128, } records = {} for line in lines: match = re.fullmatch( r"(MD5|SHA1|SHA2-256|SHA2-512)= ([0-9a-f]+)", line ) if match is None: raise ProxyError("файл SHA-256 имеет неожиданный формат") name, value = match.groups() if name in records or len(value) != expected_lengths[name]: raise ProxyError("файл SHA-256 имеет неожиданный формат") records[name] = value if set(records) != set(expected_lengths): raise ProxyError("файл SHA-256 имеет неожиданный формат") return records["SHA2-256"] def archive_sha256(archive_path): digest = hashlib.sha256() total = 0 try: with archive_path.open("rb") as stream: while True: chunk = stream.read( min(STREAM_CHUNK_BYTES, MAX_ARCHIVE_BYTES - total + 1) ) if not chunk: break total += len(chunk) if total > MAX_ARCHIVE_BYTES: raise ProxyError("архив Xray превышает допустимый размер") digest.update(chunk) except ProxyError: raise except OSError as error: raise ProxyError("не удалось прочитать архив Xray: {}".format(error)) return digest.hexdigest() def extract_xray_candidate(archive_path, candidate_path): try: with zipfile.ZipFile(str(archive_path)) as archive: members = [item for item in archive.infolist() if item.filename == "xray"] if len(members) != 1 or members[0].is_dir(): raise ProxyError("официальный архив не содержит ровно один файл xray") if members[0].file_size > MAX_XRAY_BYTES: raise ProxyError("извлекаемый xray превышает допустимый размер") with archive.open(members[0]) as source: descriptor = os.open( str(candidate_path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, ) with os.fdopen(descriptor, "wb") as destination: extracted = 0 while True: chunk = source.read( min(STREAM_CHUNK_BYTES, MAX_XRAY_BYTES - extracted + 1) ) if not chunk: break extracted += len(chunk) if extracted > MAX_XRAY_BYTES: raise ProxyError( "извлекаемый xray превышает допустимый размер" ) destination.write(chunk) if extracted != members[0].file_size: raise ProxyError("размер извлечённого xray не совпадает с ZIP") candidate_path.chmod(0o755) except ProxyError: raise except (OSError, KeyError, zipfile.BadZipFile) as error: raise ProxyError("не удалось извлечь xray из проверенного архива: {}".format(error)) def install_destination(): return Path.home() / ".local/bin/xray" def require_install_parent_within_home(parent): try: home = Path.home().resolve(strict=True) resolved_parent = parent.resolve(strict=False) resolved_parent.relative_to(home) except (OSError, RuntimeError, ValueError): raise ProxyError( "каталог установки должен находиться внутри домашнего каталога пользователя" ) def require_install_safe_managed_state(): path = state_path() try: path.lstat() except FileNotFoundError: return except OSError: raise ProxyError( "состояние managed-режима не удалось проверить; сначала выполните proxy-client stop" ) try: state = read_managed_state() outcome = managed_state_outcome(state) except ProxyError: raise ProxyError( "состояние managed-режима не подтверждено; сначала выполните proxy-client stop" ) if outcome != "stale": raise ProxyError( "managed Xray работает или не подтверждён; сначала выполните proxy-client stop" ) def acquire_install_lock(): try: return acquire_start_lock() except StartLockBusyError: raise ProxyError( "другая операция удерживает блокировку; сначала выполните proxy-client stop" ) def existing_installation_outcome(destination, replace): try: info = destination.lstat() except FileNotFoundError: return "install" except OSError as error: raise ProxyError("не удалось проверить назначение {}: {}".format(destination, error)) if stat.S_ISDIR(info.st_mode): raise ProxyError("назначение {} является каталогом".format(destination)) if replace: return "install" if stat.S_ISREG(info.st_mode) and os.access(str(destination), os.X_OK): try: xray_version(str(destination)) except ProxyError: pass else: return "already-installed" raise ProxyError( "назначение {} уже существует; для проверенной замены используйте --replace".format( destination ) ) def verify_installed_xray_on_path(destination): found = shutil.which("xray") try: matches = found is not None and Path(found).resolve(strict=True) == destination.resolve( strict=True ) except (OSError, RuntimeError): matches = False if not matches: raise ProxyError( "Xray установлен в {}, но PATH выбирает другой файл; добавьте ~/.local/bin в начало PATH".format( destination ) ) def command_install(replace=False, downloader=download_file): destination = install_destination() lock = acquire_install_lock() try: require_install_safe_managed_state() require_install_parent_within_home(destination.parent) outcome = existing_installation_outcome(destination, replace) if outcome == "already-installed": print("Xray {} уже установлен: {}".format(XRAY_VERSION, destination)) else: asset = xray_asset_for_architecture(platform.machine()) try: destination.parent.mkdir(parents=True, exist_ok=True, mode=0o755) except OSError as error: raise ProxyError( "не удалось создать каталог установки {}: {}".format( destination.parent, error ) ) require_install_parent_within_home(destination.parent) if not destination.parent.is_dir(): raise ProxyError( "каталог установки {} не является каталогом".format(destination.parent) ) staging = Path( tempfile.mkdtemp(prefix=".xray-install-", dir=str(destination.parent)) ) try: staging.chmod(0o700) archive_path = staging / asset digest_path = staging / (asset + ".dgst") candidate_path = staging / "xray" downloader( XRAY_RELEASE_BASE_URL + asset, archive_path, MAX_ARCHIVE_BYTES, ) downloader( XRAY_RELEASE_BASE_URL + asset + ".dgst", digest_path, MAX_DIGEST_BYTES, ) expected = expected_archive_sha256(digest_path) if archive_sha256(archive_path) != expected: raise ProxyError("SHA-256 архива Xray не совпадает") extract_xray_candidate(archive_path, candidate_path) xray_version(str(candidate_path)) os.replace(str(candidate_path), str(destination)) finally: shutil.rmtree(str(staging), ignore_errors=True) print("Xray {} установлен: {}".format(XRAY_VERSION, destination)) finally: lock.close() verify_installed_xray_on_path(destination) return 0 def run_xray_test(path, config_path): try: completed = subprocess.run( [path, "run", "-test", "-c", str(config_path)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10, check=False, ) except (OSError, subprocess.TimeoutExpired): raise ProxyError("проверка созданной конфигурации Xray не выполнена") if completed.returncode != 0: raise ProxyError("созданная конфигурация не прошла xray run -test") def process_alive(pid): try: Path("/proc/{}".format(pid)).stat() return True except FileNotFoundError: return False except OSError: return True def parse_proc_stat_starttime(text): comm_end = text.rfind(")") if comm_end < 0: raise ValueError("missing comm terminator") fields = text[comm_end + 1 :].split() if len(fields) <= 19: raise ValueError("missing starttime") return int(fields[19]) def process_starttime(pid): try: text = Path("/proc/{}/stat".format(pid)).read_text(encoding="ascii") return parse_proc_stat_starttime(text) except (OSError, UnicodeError, ValueError): return None def process_executable(pid): try: return str(Path("/proc/{}/exe".format(pid)).resolve(strict=True)) except (OSError, RuntimeError): return None def read_boot_id(): try: value = Path("/proc/sys/kernel/random/boot_id").read_text( encoding="ascii" ).strip() except (OSError, UnicodeError): return None return value or None def require_boot_id(): value = read_boot_id() if value is None: raise ProxyError("не удалось определить идентификатор текущей загрузки") return value def process_cmdline(pid): try: raw = Path("/proc/{}/cmdline".format(pid)).read_bytes() except OSError: return [] return [item.decode("utf-8", "surrogateescape") for item in raw.split(b"\0") if item] def process_argument_resolves_to(pid, arguments, expected_path): try: process_cwd = Path("/proc/{}/cwd".format(pid)).resolve(strict=True) except (OSError, RuntimeError): return False for argument in arguments: if not argument or "\0" in argument: continue candidate = Path(argument) if not candidate.is_absolute(): candidate = process_cwd / candidate try: if candidate.resolve(strict=True) == expected_path: return True except (OSError, RuntimeError): continue return False def read_managed_state(): path = state_path() try: info = path.lstat() if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode): raise ProxyError("состояние управляемого Xray не подтверждено") if info.st_uid != os.getuid() or info.st_mode & 0o077: raise ProxyError("состояние управляемого Xray не подтверждено") value = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: raise ProxyError("управляемый Xray не запущен; выполните proxy-client start") except ProxyError: raise except (OSError, ValueError, TypeError): raise ProxyError("состояние управляемого Xray не подтверждено") if not isinstance(value, dict) or set(value) != STATE_KEYS: raise ProxyError("состояние управляемого Xray не подтверждено") integer_keys = ( "supervisor_pid", "xray_pid", "port", "supervisor_starttime", "xray_starttime", ) if not all( isinstance(value[key], int) and not isinstance(value[key], bool) for key in integer_keys ): raise ProxyError("состояние управляемого Xray не подтверждено") string_keys = ( "xray_path", "config_path", "address", "supervisor_path", "boot_id", ) if not all(isinstance(value[key], str) for key in string_keys): raise ProxyError("состояние управляемого Xray не подтверждено") return value def supervisor_identity_matches(state): pid = state["supervisor_pid"] if pid <= 1: return False arguments = process_cmdline(pid) if len(arguments) < 2: return False if not process_argument_resolves_to( pid, [arguments[1]], Path(state["supervisor_path"]) ): return False return process_starttime(pid) == state["supervisor_starttime"] def xray_identity_matches(state): pid = state["xray_pid"] if pid <= 1 or process_executable(pid) != state["xray_path"]: return False return process_starttime(pid) == state["xray_starttime"] def wait_for_child_starttime(child): while True: value = process_starttime(child.pid) if value is not None: return value if child.poll() is not None: raise ProxyError("Xray завершился до сохранения идентичности процесса") time.sleep(0.01) def signal_supervisor(state, signal_number): if read_boot_id() != state["boot_id"] or not supervisor_identity_matches(state): return False try: os.kill(state["supervisor_pid"], signal_number) except OSError: return False return True def signal_xray(state, signal_number): if read_boot_id() != state["boot_id"] or not xray_identity_matches(state): return False try: os.kill(state["xray_pid"], signal_number) except OSError: return False return True def managed_state_outcome(state): current_boot_id = read_boot_id() if current_boot_id is None: return "unverifiable" if current_boot_id != state["boot_id"]: return "stale" if state["supervisor_pid"] == state["xray_pid"]: return "unverifiable" supervisor_alive = process_alive(state["supervisor_pid"]) xray_alive = process_alive(state["xray_pid"]) if not supervisor_alive and not xray_alive: return "stale" supervisor_matches = supervisor_identity_matches(state) xray_matches = xray_identity_matches(state) if supervisor_matches and xray_matches: return "healthy" if not supervisor_alive and xray_matches: return "orphan" return "unverifiable" def cleanup_stale_files(): for path in (state_path(), generated_config_path()): try: path.unlink() except FileNotFoundError: pass def unverifiable_state_error(): return ProxyError( "личность процесса не подтверждена; сигналы не отправлены, " "удалите state.json и client.json вручную после проверки процессов" ) def verify_managed_state(config=None, require_port=True): state = read_managed_state() outcome = managed_state_outcome(state) if outcome == "stale": lock = acquire_start_lock() try: state = read_managed_state() outcome = managed_state_outcome(state) if outcome == "stale": cleanup_stale_files() raise ProxyError("устаревшее состояние управляемого Xray очищено") finally: lock.close() if outcome == "orphan": raise ProxyError("обнаружен Xray-сирота; выполните proxy-client stop") if outcome != "healthy": raise unverifiable_state_error() if config is not None and ( config.proxy_type != "xray" or config.address != state["address"] or config.port != state["port"] ): raise ProxyError("настройки и состояние управляемого Xray не совпадают") if require_port: check_endpoint(state["address"], state["port"]) return state def acquire_start_lock(): private_directory(state_home()) try: descriptor = os.open(str(lock_path()), os.O_RDWR | os.O_CREAT, 0o600) lock = os.fdopen(descriptor, "r+") fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) os.chmod(str(lock_path()), 0o600) except BlockingIOError: if "lock" in locals(): lock.close() raise StartLockBusyError("другой управляемый запуск уже удерживает блокировку") except OSError as error: if "lock" in locals(): lock.close() elif "descriptor" in locals(): os.close(descriptor) raise ProxyError("не удалось открыть файл блокировки: {}".format(error)) return lock def safe_summary(profile, local_port): query = profile.query print( "Профиль: сервер {}:{}, транспорт {}, защита {}, SNI {}, fingerprint {}, локальный порт {}".format( profile.server, profile.port, "RAW" if profile.transport == "tcp" else "XHTTP", profile.security.upper(), query["sni"], query["fp"], local_port, ), flush=True, ) if profile.transport == "tcp": if "host" in query or "alpn" in query: print("TCP/REALITY: Host и ALPN из ссылки не используются", flush=True) print("Секреты REALITY: скрыты", flush=True) else: print( "XHTTP: Host {}, длина пути: {}, ALPN {}".format( query["host"], len(query["path"]), query.get("alpn", "не задан") ), flush=True, ) print("UUID и ключи: скрыты", flush=True) def warn_legacy(profile): if profile.legacy_insecure: sys.stderr.write( "Предупреждение: legacy insecure не переносится; сертификат проверяется обычным способом.\n" ) def diagnose_ip(config): connection = http.client.HTTPConnection( config.address, config.port, timeout=preflight_timeout() ) try: connection.request( "GET", "http://ifconfig.me/ip", headers={"Host": "ifconfig.me", "User-Agent": "proxy-client/1"}, ) response = connection.getresponse() address = response.read(256).decode("ascii", "replace").strip() if response.status != 200 or not address: raise ValueError("empty response") parsed_address = ipaddress.ip_address(address) print("Внешний IP через прокси: {}".format(parsed_address), flush=True) except (OSError, ValueError, http.client.HTTPException): sys.stderr.write( "Предупреждение: внешний IP через прокси определить не удалось; Xray продолжает работу.\n" ) finally: connection.close() def cleanup_managed_files(supervisor_pid): try: current = read_managed_state() if current["supervisor_pid"] == supervisor_pid: state_path().unlink() except ProxyError: pass try: generated_config_path().unlink() except FileNotFoundError: pass def command_status(config): if config.proxy_type == "external": check_reachable(config, preflight_timeout()) print( "{} (external) доступен по адресу {}:{}".format( config.name, config.address, config.port ) ) else: verify_managed_state(config) print( "{}: управляемый Xray работает на {}:{}".format( config.name, config.address, config.port ) ) return 0 def command_env(config): if config.proxy_type == "external": check_reachable(config, preflight_timeout()) else: verify_managed_state(config) lines = ["{}={}".format(key, value) for key, value in environment_values(config)] sys.stdout.write("\n".join(lines) + "\n") return 0 def command_run(config, remainder): if not remainder or remainder[0] != "--" or len(remainder) == 1: raise ProxyError("использование: proxy-client run -- COMMAND ARG...") command = remainder[1:] if shutil.which(command[0]) is None: sys.stderr.write("Ошибка: команда не найдена в PATH: {}\n".format(command[0])) return 127 if config.proxy_type == "external": check_reachable(config, preflight_timeout()) else: verify_managed_state(config) child_environment = os.environ.copy() child_environment.update(dict(environment_values(config))) try: completed = subprocess.run(command, env=child_environment, check=False) except FileNotFoundError: sys.stderr.write("Ошибка: команда не найдена в PATH: {}\n".format(command[0])) return 127 except OSError as error: sys.stderr.write("Ошибка: не удалось запустить {}: {}\n".format(command[0], error)) return 126 if completed.returncode < 0: return 128 - completed.returncode return completed.returncode def command_start(config, profile_file): if config.proxy_type == "external": raise ProxyError("start доступен только для type=xray; измените машинный конфиг") try: address = ipaddress.ip_address(config.address) except ValueError: raise ProxyError("для type=xray поле address должно быть loopback IP-адресом") if not address.is_loopback: raise ProxyError("для type=xray требуется loopback IP-адрес") lock = acquire_start_lock() try: return command_start_locked(config, profile_file, address) finally: lock.close() def check_port_available(address, port): family = socket.AF_INET6 if address.version == 6 else socket.AF_INET try: with socket.socket(family, socket.SOCK_STREAM) as listener: listener.bind((str(address), port)) except OSError: raise ProxyError("локальный порт {} занят или недоступен".format(port)) def command_start_locked(config, profile_file, address): if state_path().exists() or state_path().is_symlink(): try: existing_state = read_managed_state() except ProxyError: raise unverifiable_state_error() outcome = managed_state_outcome(existing_state) if outcome == "stale": cleanup_stale_files() elif outcome == "orphan": raise ProxyError("обнаружен Xray-сирота; сначала выполните proxy-client stop") elif outcome == "healthy": raise ProxyError("управляемый Xray уже запущен") else: raise unverifiable_state_error() check_port_available(address, config.port) profile_path = ( Path(profile_file) if profile_file else proxy_client_config_home() / "default.url" ) raw_profile = read_profile(profile_path) executable = resolve_xray() version = xray_version(executable) profile = parse_vless(raw_profile) generated = build_xray_config(profile, config.address, config.port) write_private_json(generated_config_path(), generated) try: run_xray_test(executable, generated_config_path()) print("Xray: {} ({})".format(executable, version), flush=True) safe_summary(profile, config.port) warn_legacy(profile) stopping = {"signal": None} def request_stop(signal_number, _frame): if stopping["signal"] is None: stopping["signal"] = signal_number previous = {} for handled in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP): previous[handled] = signal.signal(handled, request_stop) boot_id = require_boot_id() supervisor_path = str(Path(__file__).resolve(strict=True)) supervisor_starttime = process_starttime(os.getpid()) if supervisor_starttime is None: raise ProxyError("не удалось сохранить идентичность супервизора") child = subprocess.Popen( [executable, "run", "-c", str(generated_config_path())] ) xray_starttime = wait_for_child_starttime(child) state = { "supervisor_pid": os.getpid(), "xray_pid": child.pid, "xray_path": executable, "config_path": str(generated_config_path()), "address": config.address, "port": config.port, "supervisor_path": supervisor_path, "supervisor_starttime": supervisor_starttime, "xray_starttime": xray_starttime, "boot_id": boot_id, } write_private_json(state_path(), state) deadline = time.monotonic() + max(1.0, preflight_timeout()) last_error = None while time.monotonic() < deadline: if child.poll() is not None: raise ProxyError("Xray завершился до готовности локального порта") try: check_endpoint(config.address, config.port) last_error = None break except ProxyError as error: last_error = error time.sleep(0.03) if last_error is not None: raise ProxyError( "локальный порт {} занят или недоступен".format(config.port) ) diagnose_ip(config) while child.poll() is None and stopping["signal"] is None: time.sleep(0.05) if stopping["signal"] is not None and child.poll() is None: if not signal_xray(state, stopping["signal"]): if child.poll() is None: raise unverifiable_state_error() else: try: child.wait(timeout=stop_timeout()) except subprocess.TimeoutExpired: if not signal_xray(state, signal.SIGKILL): if child.poll() is None: raise unverifiable_state_error() else: child.wait() return_code = child.wait() if stopping["signal"] is not None: return 0 if return_code < 0: return 128 - return_code return return_code finally: if "previous" in locals(): for handled, handler in previous.items(): signal.signal(handled, handler) preserve_managed_files = False if "child" in locals() and child.poll() is None and "state" in locals(): if signal_xray(state, signal.SIGTERM): try: child.wait(timeout=stop_timeout()) except subprocess.TimeoutExpired: if signal_xray(state, signal.SIGKILL): child.wait() elif child.poll() is None: preserve_managed_files = True elif child.poll() is None: preserve_managed_files = True elif "child" in locals() and child.poll() is None: preserve_managed_files = True if preserve_managed_files: raise unverifiable_state_error() cleanup_managed_files(os.getpid()) def stop_timeout(): raw = os.environ.get("PROXY_CLIENT_STOP_TIMEOUT", "5") try: value = float(raw) if not value > 0: raise ValueError except ValueError: raise ProxyError("PROXY_CLIENT_STOP_TIMEOUT должен быть положительным числом секунд") return value def stop_managed_state(timeout, state, outcome): if outcome == "stale": cleanup_stale_files() raise ProxyError("устаревшее состояние управляемого Xray очищено") if outcome == "unverifiable": raise unverifiable_state_error() if outcome == "orphan": if not signal_xray(state, signal.SIGTERM): raise ProxyError("личность Xray-сироты изменилась; сигнал не отправлен") deadline = time.monotonic() + timeout while time.monotonic() < deadline: if not process_alive(state["xray_pid"]): cleanup_stale_files() print("Xray-сирота остановлен, служебные файлы удалены") return 0 time.sleep(0.05) if not signal_xray(state, signal.SIGKILL): raise ProxyError("личность Xray-сироты изменилась; SIGKILL не отправлен") deadline = time.monotonic() + timeout while time.monotonic() < deadline: if not process_alive(state["xray_pid"]): cleanup_stale_files() print("Xray-сирота остановлен, служебные файлы удалены") return 0 time.sleep(0.05) raise ProxyError("остановка Xray-сироты после SIGKILL превысила таймаут") check_endpoint(state["address"], state["port"]) if not signal_supervisor(state, signal.SIGTERM): raise ProxyError("не удалось послать сигнал подтверждённому супервизору") deadline = time.monotonic() + timeout while time.monotonic() < deadline: if not process_alive(state["xray_pid"]) and not state_path().exists(): print("Управляемый Xray остановлен") return 0 time.sleep(0.05) raise ProxyError("штатная остановка управляемого Xray превысила таймаут") def command_stop(): timeout = stop_timeout() state = read_managed_state() outcome = managed_state_outcome(state) if outcome in ("stale", "orphan"): lock = acquire_start_lock() try: state = read_managed_state() outcome = managed_state_outcome(state) return stop_managed_state(timeout, state, outcome) finally: lock.close() return stop_managed_state(timeout, state, outcome) def build_parser(): parser = argparse.ArgumentParser( prog="proxy-client", description="Безопасный запуск команд через явно настроенный прокси.", ) subparsers = parser.add_subparsers(dest="subcommand") install_parser = subparsers.add_parser("install", help="установить проверенный Xray") install_parser.add_argument( "--replace", action="store_true", help="проверенно заменить существующее назначение" ) subparsers.add_parser("status", help="проверить доступность активного прокси") subparsers.add_parser("env", help="вывести восемь proxy-переменных") start_parser = subparsers.add_parser("start", help="запустить управляемый Xray") start_parser.add_argument("-f", "--file", dest="profile_file") subparsers.add_parser("stop", help="остановить управляемый Xray") run_parser = subparsers.add_parser("run", help="запустить команду через прокси") run_parser.add_argument("arguments", nargs=argparse.REMAINDER) return parser def main(): parser = build_parser() arguments = parser.parse_args() if arguments.subcommand is None: parser.print_usage(sys.stderr) sys.stderr.write("proxy-client: ошибка: укажите явную подкоманду\n") return 2 try: if arguments.subcommand == "install": return command_install(arguments.replace) if arguments.subcommand == "stop": return command_stop() config = load_config() if arguments.subcommand == "status": return command_status(config) if arguments.subcommand == "env": return command_env(config) if arguments.subcommand == "run": return command_run(config, arguments.arguments) if arguments.subcommand == "start": return command_start(config, arguments.profile_file) except ProxyError as error: sys.stderr.write("Ошибка: {}\n".format(error)) return 1 return 2 if __name__ == "__main__": sys.exit(main())