dotfiles/proxy-client/.local/bin/proxy-client

#!/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())