#!/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 shlex
import shutil
import signal
import socket
import subprocess
import tempfile
import time
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",
    # Свой Gitea: домен пишем в двух видах, потому что программы читают список
    # по-разному — где-то запись с точкой покрывает только поддомены.
    "dementev.space",
    ".dementev.space",
)
XRAY_VERSION = "26.10.10"
VERSION_LINE = re.compile(r"Xray {}(?: .*)?".format(re.escape(XRAY_VERSION)))
XRAY_RELEASE_BASE_URL = (
    "https://github.com/XTLS/Xray-core/releases/download/v{}/".format(XRAY_VERSION)
)
# Сколько секунд ждать остановки Xray, прежде чем добить его SIGKILL.
STOP_TIMEOUT = 5
QUERY_KEYS = {
    "type",
    "security",
    "encryption",
    "flow",
    "sni",
    "fp",
    "pbk",
    "sid",
    "pqv",
    "spx",
    "ech",
    "fm",
    "support-x25519mlkem768",
    "headerType",
    "host",
    "alpn",
    "hiddify",
    "core",
    "path",
    "mode",
    "extra",
    "x_padding_bytes",
    "xPaddingBytes",
    "scMaxEachPostBytes",
    "scMinPostsIntervalMs",
    "uplinkChunkSize",
    "pcs",
    "vcn",
    "allowInsecure",
    "insecure",
}
ALPN_VALUES = {"http/1.1", "h2", "h3"}
XHTTP_RANGE_KEYS = {"xPaddingBytes", "scMaxEachPostBytes", "scMinPostsIntervalMs", "uplinkChunkSize"}


class ProxyError(Exception):
    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, insecure, finalmask=None):
        self.server = server
        self.port = port
        self.profile_id = profile_id
        self.query = query
        self.alpn = alpn
        self.extra = extra
        self.insecure = insecure
        self.finalmask = finalmask

    @property
    def transport(self):
        return self.query["type"]

    @property
    def security(self):
        return self.query["security"]


def config_home():
    root = os.environ.get("XDG_CONFIG_HOME")
    return (Path(root).expanduser() if root else Path.home() / ".config") / "proxy-client"


def state_home():
    root = os.environ.get("XDG_STATE_HOME")
    base = Path(root).expanduser() if root else Path.home() / ".local/state"
    return base / "proxy-client"


def state_path():
    return state_home() / "state.json"


def generated_config_path():
    return state_home() / "client.json"


def validate_address(value):
    try:
        ipaddress.ip_address(value)
    except ValueError:
        pass
    else:
        if "%" in value:
            raise ProxyError("поле address не поддерживает IPv6 zone ID")
        return
    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-именем")


def load_config():
    path = config_home() / "config"
    if not path.exists():
        if path.is_symlink():
            raise ProxyError("машинный конфиг {} ссылается на отсутствующий файл".format(path))
        return ProxyConfig(
            "external", DEFAULT_NAME, DEFAULT_ADDRESS, DEFAULT_PORT, DEFAULT_NO_PROXY_EXTRA
        )
    try:
        return parse_config(path.read_text(encoding="utf-8"))
    except (OSError, ValueError, configparser.Error, ProxyError) as error:
        raise ProxyError("машинный конфиг {}: {}".format(path, error))


def parse_config(text):
    parser = configparser.ConfigParser(interpolation=None, empty_lines_in_values=False)
    parser.read_string(text, source="config")
    if parser.sections() != ["proxy"]:
        raise ProxyError("нужна ровно одна секция [proxy]")
    section = parser["proxy"]
    keys = set(section)
    if keys != CONFIG_KEYS:
        details = []
        if CONFIG_KEYS - keys:
            details.append("нет ключей: {}".format(", ".join(sorted(CONFIG_KEYS - keys))))
        if keys - CONFIG_KEYS:
            details.append("неизвестные ключи: {}".format(", ".join(sorted(keys - CONFIG_KEYS))))
        raise ProxyError("неверные ключи [proxy] ({})".format("; ".join(details)))

    proxy_type = section["type"]
    if proxy_type not in ("external", "xray"):
        raise ProxyError("поле type должно быть external или xray")
    if not section["name"]:
        raise ProxyError("поле name не должно быть пустым")
    address = section["address"]
    validate_address(address)
    if not re.fullmatch(r"[0-9]+", section["port"]) or not 1 <= int(section["port"]) <= 65535:
        raise ProxyError("поле port должно быть целым числом от 1 до 65535")
    no_proxy_extra = ()
    if section["no_proxy_extra"]:
        no_proxy_extra = tuple(item.strip() for item in section["no_proxy_extra"].split(","))
        if not all(no_proxy_extra):
            raise ProxyError("no_proxy_extra содержит пустой элемент")
    return ProxyConfig(proxy_type, section["name"], address, int(section["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):
    try:
        socket.create_connection((config.address, config.port), timeout=preflight_timeout()).close()
    except OSError as error:
        raise ProxyError(
            "прокси {} ({}:{}) недоступен: {}".format(
                config.name, config.address, config.port, error
            )
        )


def check_proxy(config):
    if config.proxy_type == "xray":
        require_supervised_xray(config)
    check_reachable(config)


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)
    path.chmod(0o700)


def write_private_json(path, value):
    private_directory(path.parent)
    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)
        stream.write("\n")
    path.chmod(0o600)


def load_profile(path):
    # Права проверяем у конечного файла: у симлинка они ничего не значат.
    resolved = path.expanduser().resolve(strict=True)
    if resolved.stat().st_mode & 0o077:
        raise ProxyError(
            "файл профиля {} доступен другим пользователям; выполните: chmod 600 {}".format(
                resolved, shlex.quote(str(resolved))
            )
        )
    lines = [line.strip() for line in resolved.read_text(encoding="utf-8").splitlines()]
    lines = [line for line in lines if line]
    if len(lines) != 1:
        raise ProxyError(
            "файл профиля {} должен содержать ровно одну непустую VLESS-ссылку".format(resolved)
        )
    try:
        return parse_vless(lines[0])
    except ProxyError as error:
        raise ProxyError("файл профиля {}: {}".format(resolved, error))


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):
    if not all(query.get(key) for key in keys):
        raise ProxyError("в профиле отсутствуют обязательные параметры")


def list_value(value):
    items = value.split(",")
    if not all(items):
        raise ProxyError("списочный параметр содержит пустой элемент")
    return items


def json_object(value, name):
    try:
        result = json.loads(value)
    except ValueError:
        result = None
    if not isinstance(result, dict):
        raise ProxyError("{} должен быть JSON-объектом".format(name))
    return result


def normalize_xhttp_settings(settings):
    result = dict(settings)
    # Xray 26.9.30 переименовал поля; старые имена ядро молча игнорирует.
    for old, new in (("sessionPlacement", "sessionIDPlacement"), ("sessionKey", "sessionIDKey")):
        if old in result:
            value = result.pop(old)
            if new in result and result[new] != value:
                raise ProxyError("противоречивые старые и новые параметры сессии XHTTP")
            result[new] = value
    if isinstance(result.get("extra"), dict):
        result["extra"] = normalize_xhttp_settings(result["extra"])
    if isinstance(result.get("downloadSettings"), dict):
        download = dict(result["downloadSettings"])
        for name in ("xhttpSettings", "splithttpSettings"):
            if isinstance(download.get(name), dict):
                download[name] = normalize_xhttp_settings(download[name])
        result["downloadSettings"] = download
    return result


def parse_vless(url):
    try:
        parsed = urlsplit(url)
    except ValueError:
        raise ProxyError("неверная VLESS-ссылка")
    # Новые ссылки Hiddify пишут «/» перед «?»: ...@SERVER:PORT/?type=...
    if (
        parsed.scheme != "vless"
        or not parsed.netloc
        or parsed.password is not None
        or parsed.path not in ("", "/")
    ):
        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)
    query.setdefault("encryption", "none")
    query.setdefault("sni", server)
    if query.get("security") == "tls":
        query.setdefault("fp", "chrome")
    if query.get("type") == "raw":
        query["type"] = "tcp"
    if query.get("flow") == "":
        del query["flow"]
    required(query, "type", "security", "encryption", "sni", "fp")
    # Кроме none, Xray принимает VLESS-шифрование вида
    # mlkem768x25519plus.РЕЖИМ.RTT.КЛЮЧ...; подробности проверит xray run -test.
    encrypted = query["encryption"] != "none"
    if encrypted and not query["encryption"].startswith("mlkem768x25519plus."):
        raise ProxyError("поддерживается только encryption=none или mlkem768x25519plus")
    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")
    if "support-x25519mlkem768" in query and query["support-x25519mlkem768"] != "true":
        raise ProxyError("поддерживается только метаданные support-x25519mlkem768=true")
    # Xray 26.10.10 больше не умеет отключать проверку сертификата: поля
    # принимаем, но не переносим.
    insecure = any(
        query.get(key, "").lower() in ("1", "true") for key in ("allowInsecure", "insecure")
    )
    alpn = None
    extra = None
    finalmask = json_object(query["fm"], "fm") if "fm" in query else None

    if query["type"] == "tcp" and query["security"] == "reality":
        required(query, "pbk")
        if ({"path", "mode", "extra", "x_padding_bytes", "pcs", "vcn", "core", "ech"}
                | XHTTP_RANGE_KEYS).intersection(query):
            raise ProxyError("параметры профиля несовместимы с TCP/REALITY")
    elif query["type"] == "xhttp" and query["security"] == "tls":
        query.setdefault("path", "/")
        required(query, "path")
        # Vision поверх XHTTP работает только внутри VLESS-шифрования.
        if (("flow" in query and not encrypted)
                or {"pbk", "sid", "pqv", "spx", "support-x25519mlkem768"}.intersection(query)):
            raise ProxyError("параметры профиля несовместимы с XHTTP/TLS")
        if query.get("pcs"):
            hashes = list_value(query["pcs"])
            if any(re.fullmatch(r"[0-9A-Fa-f]{64}", item.strip().replace(":", ""))
                   is None for item in hashes):
                raise ProxyError("pcs должен содержать 32-байтные SHA-256 в hex")
        if query.get("vcn"):
            list_value(query["vcn"])
        if "alpn" in query:
            alpn = list_value(query["alpn"])
            if any(item not in ALPN_VALUES for item in alpn):
                raise ProxyError("профиль содержит неподдерживаемый ALPN")
        if "extra" in query:
            extra = json_object(query["extra"], "extra")
        if "x_padding_bytes" in query:
            if extra is None:
                extra = {}
            # При наличии extra Xray берёт padding только из него.
            extra.setdefault("xPaddingBytes", query["x_padding_bytes"])
        for key in XHTTP_RANGE_KEYS.intersection(query):
            if extra is None:
                extra = {}
            extra[key] = query[key]
        if extra is not None:
            extra = normalize_xhttp_settings(extra)
    else:
        raise ProxyError("поддерживаются только TCP/REALITY и XHTTP/TLS")

    return VlessProfile(server, port, profile_id, query, alpn, extra, insecure, finalmask)


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.get("sid", ""),
        }
        if "pqv" in query:
            reality["mldsa65Verify"] = query["pqv"]
        if "spx" in query:
            reality["spiderX"] = query["spx"]
        stream = {
            "network": "raw",
            "security": "reality",
            "realitySettings": reality,
        }
    else:
        xhttp = {"host": query.get("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"]
        if "ech" in query:
            tls["echConfigList"] = query["ech"]
        stream = {
            "network": "xhttp",
            "security": "tls",
            "xhttpSettings": xhttp,
            "tlsSettings": tls,
        }
    if profile.finalmask is not None:
        stream["finalmask"] = profile.finalmask
    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")
    return str(Path(found).resolve(strict=True))


def check_xray_version(path):
    try:
        completed = subprocess.run(
            [path, "version"],
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            timeout=5,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        completed = None
    lines = completed.stdout.splitlines() if completed else []
    if not lines or completed.returncode != 0 or not VERSION_LINE.fullmatch(lines[0]):
        raise ProxyError("{} не сообщил версию Xray {}".format(path, XRAY_VERSION))


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):
    try:
        with urllib.request.urlopen(url, timeout=30) as response:
            with destination.open("wb") as stream:
                shutil.copyfileobj(response, stream)
    except (OSError, http.client.HTTPException) as error:
        raise ProxyError("не удалось скачать {}: {}".format(url, error))


def expected_sha256(digest_text):
    for line in digest_text.splitlines():
        name, _, value = line.partition("=")
        if name.strip() == "SHA2-256":
            return value.strip().lower()
    raise ProxyError("в файле .dgst нет строки SHA2-256=")


def install_xray(destination, downloader):
    asset = xray_asset_for_architecture(platform.machine())
    destination.parent.mkdir(parents=True, exist_ok=True)
    # Временный каталог рядом с назначением: os.replace тогда атомарен, а
    # неудачная установка не трогает старый бинарник.
    staging = Path(tempfile.mkdtemp(prefix=".xray-install-", dir=str(destination.parent)))
    try:
        archive = staging / asset
        digest = staging / (asset + ".dgst")
        downloader(XRAY_RELEASE_BASE_URL + asset, archive)
        downloader(XRAY_RELEASE_BASE_URL + asset + ".dgst", digest)
        expected = expected_sha256(digest.read_text(encoding="ascii", errors="replace"))
        if hashlib.sha256(archive.read_bytes()).hexdigest() != expected:
            raise ProxyError("SHA-256 архива {} не совпадает с {}.dgst".format(asset, asset))
        try:
            with zipfile.ZipFile(str(archive)) as bundle:
                candidate = Path(bundle.extract("xray", str(staging)))
        except (KeyError, zipfile.BadZipFile) as error:
            raise ProxyError("не удалось извлечь xray из {}: {}".format(asset, error))
        candidate.chmod(0o755)
        check_xray_version(str(candidate))
        os.replace(str(candidate), str(destination))
    finally:
        shutil.rmtree(str(staging), ignore_errors=True)


def command_install(replace=False, downloader=download_file):
    destination = Path.home() / ".local/bin/xray"
    if destination.is_dir():
        raise ProxyError("назначение {} является каталогом".format(destination))
    if os.path.lexists(str(destination)) and not replace:
        try:
            check_xray_version(str(destination))
        except ProxyError:
            raise ProxyError(
                "{} уже существует и это не Xray {}; заменить: proxy-client install --replace".format(
                    destination, XRAY_VERSION
                )
            )
        print("Xray {} уже установлен: {}".format(XRAY_VERSION, destination))
    else:
        install_xray(destination, downloader)
        print("Xray {} установлен: {}".format(XRAY_VERSION, destination))
        if managed_xray_running():
            print("Работающий Xray остался прежним; перезапустите proxy-client start")
    found = shutil.which("xray")
    if found is None or Path(found).resolve() != destination.resolve():
        raise ProxyError(
            "Xray установлен в {}, но PATH выбирает другой xray или не находит его; "
            'выполните: export PATH="$HOME/.local/bin:$PATH"'.format(destination)
        )
    return 0


def run_xray_test(path, config_path):
    # Вывод Xray глушим: он может процитировать секреты из конфига.
    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):
        completed = None
    if completed is None or completed.returncode != 0:
        raise ProxyError("созданная конфигурация не прошла xray run -test")


def read_state():
    path = state_path()
    try:
        state = json.loads(path.read_text(encoding="utf-8"))
        keys = ("supervisor_pid", "xray_pid", "xray_path", "address", "port")
        return {key: state[key] for key in keys}
    except FileNotFoundError:
        raise ProxyError("управляемый Xray не запущен; выполните proxy-client start")
    except (ValueError, TypeError, KeyError):
        raise ProxyError(
            "файл состояния {} повреждён; если Xray не запущен, выполните: rm {}".format(
                path, shlex.quote(str(path))
            )
        )


def xray_running(state):
    # Процесс с этим PID должен исполнять тот самый xray. После
    # install --replace ядро дописывает к пути старого файла « (deleted)».
    try:
        executable = os.readlink("/proc/{}/exe".format(state["xray_pid"]))
    except OSError:
        return False
    return executable in (state["xray_path"], state["xray_path"] + " (deleted)")


def supervisor_running(state):
    try:
        cmdline = Path("/proc/{}/cmdline".format(state["supervisor_pid"])).read_bytes()
    except OSError:
        return False
    return b"proxy-client" in cmdline


def remove_state_files():
    for path in (state_path(), generated_config_path()):
        try:
            path.unlink()
        except FileNotFoundError:
            pass


def managed_state():
    """Состояние работающего Xray; устаревшие служебные файлы удаляет."""
    state = read_state()
    if not xray_running(state):
        remove_state_files()
        raise ProxyError("управляемый Xray не запущен; устаревшие служебные файлы удалены")
    return state


def managed_xray_running():
    try:
        return xray_running(read_state())
    except ProxyError:
        return False


def require_supervised_xray(config):
    state = managed_state()
    if not supervisor_running(state):
        raise ProxyError(
            "Xray работает без proxy-client start (его процесс завершён); "
            "выполните proxy-client stop"
        )
    if (config.address, config.port) != (state["address"], state["port"]):
        raise ProxyError(
            "Xray запущен на {}:{}, а {} указывает {}:{}; чтобы применить конфиг, "
            "выполните proxy-client stop, затем proxy-client start".format(
                state["address"], state["port"], config_home() / "config",
                config.address, config.port,
            )
        )


def wait_until(condition, timeout):
    deadline = time.monotonic() + timeout
    while not condition():
        if time.monotonic() > deadline:
            return False
        time.sleep(0.05)
    return True


def port_open(address, port):
    try:
        socket.create_connection((address, port), timeout=0.2).close()
    except OSError:
        return False
    return True


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:
            # Без флага bind отказывает, пока после остановки Xray на порту
            # доживают закрытые соединения (TIME_WAIT); живой слушатель
            # на Linux флаг не обходит, и занятый порт по-прежнему виден.
            listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            listener.bind((str(address), port))
    except OSError:
        raise ProxyError("локальный порт {} занят или недоступен".format(port))


def acquire_start_lock():
    private_directory(state_home())
    lock = (state_home() / "lock").open("a")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        lock.close()
        raise ProxyError("управляемый Xray уже запущен другим proxy-client start")
    return lock


def print_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,
        )
    )
    if profile.transport == "tcp":
        if "host" in query or "alpn" in query:
            print("TCP/REALITY: Host и ALPN из ссылки не используются")
        print("Секреты REALITY: скрыты")
    else:
        print(
            "XHTTP: Host {}, длина пути: {}, ALPN {}".format(
                query.get("host") or "не задан", len(query["path"]), query.get("alpn", "не задан")
            )
        )
    if query["encryption"] != "none":
        print("VLESS-шифрование: mlkem768x25519plus")
    print("UUID и ключи: скрыты", flush=True)
    if profile.insecure:
        sys.stderr.write(
            "Предупреждение: allowInsecure/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:
            raise ValueError("unexpected status")
        print("Внешний IP через прокси: {}".format(ipaddress.ip_address(address)), flush=True)
    except (OSError, ValueError, http.client.HTTPException):
        sys.stderr.write(
            "Предупреждение: внешний IP через прокси определить не удалось; Xray продолжает работу.\n"
        )
    finally:
        connection.close()


def exit_code(returncode):
    # Как в shell: убитый сигналом N процесс даёт 128+N.
    return 128 - returncode if returncode < 0 else returncode


def interrupt(_signum, _frame):
    raise KeyboardInterrupt


def command_status(config):
    check_proxy(config)
    if config.proxy_type == "external":
        print("{} (external) доступен по адресу {}:{}".format(config.name, config.address, config.port))
    else:
        print("{}: управляемый Xray работает на {}:{}".format(config.name, config.address, config.port))
    return 0


def command_env(config):
    check_proxy(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
    check_proxy(config)
    child_environment = os.environ.copy()
    child_environment.update(dict(environment_values(config)))
    try:
        completed = subprocess.run(command, env=child_environment, check=False)
    except OSError as error:
        sys.stderr.write("Ошибка: не удалось запустить {}: {}\n".format(command[0], error))
        return 126
    return exit_code(completed.returncode)


def command_start(config, profile_file):
    if config.proxy_type != "xray":
        raise ProxyError(
            "start доступен только для type=xray; измените {}".format(config_home() / "config")
        )
    try:
        loopback = ipaddress.ip_address(config.address).is_loopback
    except ValueError:
        loopback = False
    if not loopback:
        raise ProxyError("для type=xray поле address должно быть loopback-адресом, например 127.0.0.1")
    lock = acquire_start_lock()
    try:
        return run_managed_xray(config, profile_file)
    finally:
        lock.close()


def run_managed_xray(config, profile_file):
    if state_path().exists():
        if xray_running(read_state()):
            raise ProxyError(
                "Xray остался от прошлого proxy-client start; сначала выполните proxy-client stop"
            )
        remove_state_files()
    check_port_available(ipaddress.ip_address(config.address), config.port)
    profile_path = Path(profile_file) if profile_file else config_home() / "default.url"
    profile = load_profile(profile_path)
    executable = resolve_xray()
    check_xray_version(executable)
    write_private_json(
        generated_config_path(), build_xray_config(profile, config.address, config.port)
    )
    child = None
    try:
        run_xray_test(executable, generated_config_path())
        print("Xray: {} ({})".format(executable, XRAY_VERSION))
        print_summary(profile, config.port)
        # SIGTERM (от proxy-client stop) и SIGHUP обрабатываем как Ctrl+C.
        signal.signal(signal.SIGTERM, interrupt)
        signal.signal(signal.SIGHUP, interrupt)
        child = subprocess.Popen([executable, "run", "-c", str(generated_config_path())])
        write_private_json(
            state_path(),
            {
                "supervisor_pid": os.getpid(),
                "xray_pid": child.pid,
                "xray_path": executable,
                "address": config.address,
                "port": config.port,
            },
        )
        if not wait_until(
            lambda: port_open(config.address, config.port), max(1.0, preflight_timeout())
        ):
            raise ProxyError("Xray не открыл локальный порт {}".format(config.port))
        diagnose_ip(config)
        print(
            "Управляемый Xray работает на {}:{}; остановка: Ctrl+C или proxy-client stop".format(
                config.address, config.port
            ),
            flush=True,
        )
        return exit_code(child.wait())
    except KeyboardInterrupt:
        return 0
    finally:
        if child is not None and child.poll() is None:
            child.terminate()
            try:
                child.wait(STOP_TIMEOUT)
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait()
        remove_state_files()


def command_stop():
    state = managed_state()
    # Живой супервизор сам остановит Xray; иначе Xray гасим напрямую.
    running_supervisor = supervisor_running(state)
    target = state["supervisor_pid"] if running_supervisor else state["xray_pid"]
    os.kill(target, signal.SIGTERM)

    def stopped():
        return not xray_running(state) and not supervisor_running(state)

    if not wait_until(stopped, STOP_TIMEOUT + 1):
        if xray_running(state):
            os.kill(state["xray_pid"], signal.SIGKILL)
        if not wait_until(stopped, STOP_TIMEOUT):
            raise ProxyError("управляемый Xray не остановился за {} с".format(2 * STOP_TIMEOUT + 1))
    remove_state_files()
    print("Управляемый Xray остановлен")
    return 0


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="заменить существующий ~/.local/bin/xray"
    )
    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, OSError) as error:
        sys.stderr.write("Ошибка: {}\n".format(error))
        return 1
    return 2


if __name__ == "__main__":
    sys.exit(main())
