#!/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 getpass
import http.client
import io
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 warnings
import zipfile
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
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"]


@dataclass(frozen=True)
class RuntimeContext:
    """Размещение и правила выбранного запуска; создаётся до выполнения команды."""

    config_dir: Path
    state_dir: Path
    xray_path: Optional[Path]
    profile_forbidden_mode: int
    profile_repair: str
    xray_output: Optional[int]
    stop_hint: str

    @property
    def state_path(self):
        return self.state_dir / "state.json"

    @property
    def generated_config_path(self):
        return self.state_dir / "client.json"


def user_context(profile="default", home=None):
    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}", profile):
        raise ProxyError("недопустимое имя профиля")
    config_root = os.environ.get("XDG_CONFIG_HOME")
    state_root = os.environ.get("XDG_STATE_HOME")
    home = home if home is not None else Path.home()
    config_dir = (Path(config_root).expanduser() if config_root else home / ".config") / "proxy-client"
    state_dir = (Path(state_root).expanduser() if state_root else home / ".local/state") / "proxy-client"
    stop_hint = "Ctrl+C или proxy-client stop"
    if profile != "default":
        config_dir = config_dir / "profiles" / profile
        state_dir = state_dir / "profiles" / profile
        if not (config_dir / "config").is_file():
            raise ProxyError("неизвестный профиль: {}".format(profile))
        stop_hint = "Ctrl+C или proxy-client --profile {} stop".format(profile)
    return RuntimeContext(config_dir, state_dir, None, 0o077,
                          "chmod 600 {0}", None, stop_hint)


def service_context(arguments):
    config_dir = Path(arguments.config_dir)
    state_dir = Path(arguments.runtime_dir)
    xray_path = Path(arguments.xray)
    if not all(path.is_absolute() for path in (config_dir, state_dir, xray_path)):
        raise ProxyError("системные пути должны быть абсолютными")
    if not (config_dir / "config").is_file():
        raise ProxyError("нет системного config")
    return RuntimeContext(config_dir, state_dir, xray_path, 0o037,
                          "sudo chown root:proxy-client {0}; sudo chmod 640 {0}",
                          subprocess.DEVNULL, "sudo systemctl stop proxy-client")


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(context):
    path = context.config_dir / "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)
    try:
        parser.read_string(text, source="config")
    except configparser.Error:
        raise ProxyError("неверный формат config; ожидается секция [proxy] с параметрами") from None
    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, context):
    if config.proxy_type == "xray":
        require_supervised_xray(config, context)
    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, context):
    # Права проверяем у конечного файла: у симлинка они ничего не значат.
    resolved = path.expanduser().resolve(strict=True)
    if resolved.stat().st_mode & context.profile_forbidden_mode:
        target = shlex.quote(str(resolved))
        repair = context.profile_repair.format(target)
        raise ProxyError(
            "файл профиля {} доступен другим пользователям; выполните: {}".format(
                resolved, repair
            )
        )
    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(context):
    found = str(context.xray_path) if context.xray_path is not None else 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, context=None):
    context = context if context is not None else user_context()
    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(context):
            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(context):
    path = context.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(context):
    for path in (context.state_path, context.generated_config_path):
        try:
            path.unlink()
        except FileNotFoundError:
            pass


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


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


def require_supervised_xray(config, context):
    state = managed_state(context)
    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"], context.config_dir / "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(context):
    private_directory(context.state_dir)
    lock = (context.state_dir / "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, context):
    check_proxy(config, context)
    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, context):
    check_proxy(config, context)
    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, context):
    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, context)
    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, context):
    if config.proxy_type != "xray":
        raise ProxyError(
            "start доступен только для type=xray; измените {}".format(context.config_dir / "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(context)
    try:
        return run_managed_xray(config, profile_file, context)
    finally:
        lock.close()


def run_managed_xray(config, profile_file, context):
    if context.state_path.exists():
        if xray_running(read_state(context)):
            raise ProxyError(
                "Xray остался от прошлого proxy-client start; сначала выполните proxy-client stop"
            )
        remove_state_files(context)
    check_port_available(ipaddress.ip_address(config.address), config.port)
    profile_path = Path(profile_file) if profile_file else context.config_dir / "default.url"
    profile = load_profile(profile_path, context)
    executable = resolve_xray(context)
    check_xray_version(executable)
    write_private_json(
        context.generated_config_path, build_xray_config(profile, config.address, config.port)
    )
    child = None
    try:
        run_xray_test(executable, context.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(context.generated_config_path)],
            stdout=context.xray_output,
            stderr=context.xray_output,
        )
        write_private_json(
            context.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 работает на {}:{}; остановка: {}".format(
                config.address, config.port, context.stop_hint
            ),
            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(context)


def command_stop(context):
    state = managed_state(context)
    # Живой супервизор сам остановит 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(context)
    print("Управляемый Xray остановлен")
    return 0


SERVICE_UNIT = """# Managed by proxy-client
[Unit]
Description=Proxy client (Xray)
After=network.target
StartLimitIntervalSec=0

[Service]
Type=simple
User=proxy-client
Group=proxy-client
RuntimeDirectory=proxy-client
RuntimeDirectoryMode=0700
UMask=0077
ExecStart=/usr/local/lib/proxy-client/proxy-client service run
Restart=on-failure
RestartSec=5
TimeoutStopSec=15
StandardOutput=journal
StandardError=journal
NoNewPrivileges=yes
ProtectSystem=strict
ProtectHome=yes
PrivateTmp=yes
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6

[Install]
WantedBy=multi-user.target
"""


def report_journal_storage(root):
    # Приоритет каталогов соответствует journald.conf; одноимённый drop-in
    # из /etc перекрывает поставляемый системой файл.
    directories = ["usr/lib/systemd", "usr/local/lib/systemd", "run/systemd", "etc/systemd"]
    main = next((root / directory / "journald.conf"
                 for directory in reversed(directories)
                 if (root / directory / "journald.conf").exists()), None)
    dropins = {}
    for directory in directories:
        for path in (root / directory / "journald.conf.d").glob("*.conf"):
            dropins[path.name] = path
    storage = None
    for path in ([main] if main else []) + [dropins[name] for name in sorted(dropins)]:
        section = ""
        for line in path.read_text().splitlines():
            line = line.strip()
            if line.startswith("["):
                section = line
            elif section == "[Journal]" and re.match(r"Storage\s*=", line):
                storage = line.split("=", 1)[1].strip() or None
    if storage is None:
        print("journald: Storage не задан; сохранение журнала зависит от сборки systemd. "
              "Проверьте journalctl --list-boots после перезагрузки.")
        return
    persistent = storage == "persistent" or (
        storage == "auto" and (root / "var/log/journal").is_dir())
    print("journald: Storage={}; журнал {} между загрузками (по конфигурации).".format(
        storage, "сохраняется" if persistent else "не сохраняется"))


def make_install_directory(path):
    if path.is_dir():
        return
    make_install_directory(path.parent)
    path.mkdir()
    path.chmod(0o755)


def install_file(target, data, mode):
    make_install_directory(target.parent)
    descriptor, temporary = tempfile.mkstemp(prefix=".install-", dir=str(target.parent))
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(data)
            os.fchmod(stream.fileno(), mode)
        os.replace(temporary, str(target))
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)


def service_config(text):
    config = parse_config(text)
    try:
        loopback = ipaddress.ip_address(config.address).is_loopback
    except ValueError:
        loopback = False
    if config.proxy_type != "xray" or not loopback:
        raise ProxyError("службе нужен type=xray и loopback-адрес")
    return config


def validate_service_source(source):
    missing = [name for name in ("config", "default.url") if not (source / name).is_file()]
    if missing:
        raise ProxyError("не хватает настроек {} в {}; запустите proxy-client service install "
                         "в терминале для настройки или укажите готовый --config-dir".format(
                             ", ".join(missing), source))
    config = service_config((source / "config").read_text(encoding="utf-8"))
    load_profile(source / "default.url", user_context())
    return config


@dataclass(frozen=True)
class ServiceSetupPlan:
    """Снимок перехода; config_text=None означает обновление без изменения HOME."""
    context: RuntimeContext
    original_config: Optional[bytes]
    config_text: Optional[str]
    profile_text: Optional[str]
    manual_state: Optional[dict] = None
    resume: bool = False
    recovery_file: Optional[str] = None
    source_profile: Optional[str] = None


def service_is_installed(root):
    manifest = root / "usr/local/lib/proxy-client/installed.json"
    return manifest.is_file() or manifest.with_name("installed.pending.json").is_file()


def render_machine_config(config, proxy_type):
    parser = configparser.ConfigParser(interpolation=None)
    parser["proxy"] = {"type": proxy_type, "name": config.name,
                       "address": config.address, "port": str(config.port),
                       "no_proxy_extra": ",".join(config.no_proxy_extra)}
    output = io.StringIO()
    parser.write(output)
    return output.getvalue()


def setup_reset_hint(context):
    snapshot = context.state_dir / "service-setup.json"
    return ("Чтобы начать заново, удалите только снимок: rm -- {}\n"
            "Системные и домашние настройки эта команда не меняет.").format(shlex.quote(str(snapshot)))


def check_setup_source(plan):
    path = Path(plan.source_profile) if plan.source_profile else plan.context.config_dir / "default.url"
    if plan.source_profile:
        changed = not path.is_file() or path.read_text(encoding="utf-8").strip() != plan.profile_text
        if not changed:
            load_profile(path, plan.context)
    else:
        changed = path.exists() or path.is_symlink()
    if changed:
        raise ProxyError("исходный файл VLESS-ссылки изменился; переход остановлен. " +
                         setup_reset_hint(plan.context))


def prepare_service_setup(args):
    context = user_context(args.profile)
    pending = context.state_dir / "service-setup.json"
    if pending.exists():
        if pending.is_symlink() or pending.stat().st_mode & 0o077:
            raise ProxyError("снимок перехода должен быть обычным закрытым файлом: {}".format(pending))
        if args.profile_file or args.port is not None:
            raise ProxyError("есть незавершённый переход; повторите без --file и --port. " +
                             setup_reset_hint(context))
        try:
            data = json.loads(pending.read_text(encoding="utf-8"))
            original = data["original_config"].encode() if data["original_config"] is not None else None
            service_config(data["config_text"])
            parse_vless(data["profile_text"])
            source_profile = data["source_profile"]
            if source_profile is not None and not isinstance(source_profile, str):
                raise ValueError("неверный путь источника")
        except (KeyError, TypeError, ValueError, AttributeError, ProxyError, configparser.Error):
            raise ProxyError("снимок незавершённого перехода повреждён или устарел: {}. ".format(pending) +
                             setup_reset_hint(context))
        manual_state = read_state(context) if managed_xray_running(context) else None
        plan = ServiceSetupPlan(context, original, data["config_text"], data["profile_text"],
                                manual_state, True, data.get("recovery_file"), source_profile)
        check_setup_source(plan)
        return plan
    config_path = context.config_dir / "config"
    original = config_path.read_bytes() if config_path.exists() else None
    config = parse_config(original.decode("utf-8")) if original is not None else None
    # Установленные файлы ещё не означают завершённый переход из ручного режима.
    if (Path("/") / "usr/local/lib/proxy-client/installed.json").is_file() and (
            config is None or config.proxy_type != "xray"):
        if args.profile_file or args.port is not None:
            raise ProxyError("настройки службы уже существуют; измените /etc/proxy-client, "
                             "обновление сохраняет их")
        return ServiceSetupPlan(context, None, None, None)
    if config_path.is_symlink():
        raise ProxyError("домашний config является симлинком; используйте --install-only")
    profile_path = Path(args.profile_file).expanduser() if args.profile_file else context.config_dir / "default.url"
    if profile_path.is_symlink() and not profile_path.exists():
        raise ProxyError("{} — битый симлинк; исправьте ссылку или выберите другой --file".format(profile_path))
    source_profile = None
    if profile_path.is_file():
        load_profile(profile_path, context)
        profile_text = profile_path.read_text(encoding="utf-8").strip()
        source_profile = str(profile_path.absolute())
    elif args.profile_file or not sys.stdin.isatty():
        raise ProxyError("нет VLESS-ссылки; укажите закрытый файл через --file или запустите в терминале")
    else:
        # getpass не должен переходить к видимому вводу, если терминал недоступен.
        with warnings.catch_warnings():
            warnings.simplefilter("error", getpass.GetPassWarning)
            try:
                profile_text = getpass.getpass("VLESS-ссылка (ввод скрыт): ").strip()
            except getpass.GetPassWarning:
                raise ProxyError("скрытый ввод недоступен; передайте ссылку через закрытый --file")
        parse_vless(profile_text)
    if config is None or config.proxy_type != "xray":
        port = DEFAULT_PORT if args.port is None else args.port
        if args.port is None and not args.yes:
            raw = input("Локальный порт [{}]: ".format(DEFAULT_PORT)).strip()
            if raw:
                try:
                    port = int(raw)
                except ValueError:
                    raise ProxyError("порт должен быть числом от 1 до 65535")
        config = ProxyConfig("xray", "Xray", "127.0.0.1", port,
                             config.no_proxy_extra if config else DEFAULT_NO_PROXY_EXTRA)
    elif args.port is not None:
        config.port = args.port
    text = render_machine_config(config, "xray")
    service_config(text)
    manual_state = read_state(context) if managed_xray_running(context) else None
    return ServiceSetupPlan(context, original, text, profile_text, manual_state,
                            recovery_file=str(profile_path.resolve()) if args.profile_file else None,
                            source_profile=source_profile)


def find_service_xray(explicit, home, root):
    if explicit:
        path = Path(explicit)
        if not path.is_absolute():
            raise ProxyError("xray должен быть абсолютным путём")
        return path
    local = home / ".local/bin/xray"
    if local.is_file():
        return local
    found = shutil.which("xray")
    if found:
        return Path(found).resolve()
    installed = root / "usr/local/lib/proxy-client/xray"
    return installed if installed.is_file() else None


def service_systemctl(*arguments, check=True):
    return subprocess.run(["sudo", "--", "systemctl", *arguments], check=check)


def service_ready(config):
    return subprocess.run(["systemctl", "is-active", "--quiet", "proxy-client.service"],
                          check=False).returncode == 0 and port_open(config.address, config.port)


def switch_home_to_service(plan):
    path = plan.context.config_dir / "config"
    config = service_config(plan.config_text)
    replacement = render_machine_config(config, "external").encode()
    current = path.read_bytes() if path.exists() else None
    if not path.is_symlink() and current == replacement:
        return
    if path.is_symlink() or current != plan.original_config:
        raise ProxyError("домашний config изменился во время установки; он оставлен без изменений. " +
                         setup_reset_hint(plan.context))
    private_directory(plan.context.config_dir)
    if plan.original_config is not None:
        descriptor, backup = tempfile.mkstemp(prefix="config.before-service.", dir=str(path.parent))
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(plan.original_config)
        print("Прежний config сохранён: {}".format(backup))
    install_file(path, replacement, 0o600)


def apply_service_setup(args, plan, xray):
    if plan.config_text is None:
        with tempfile.TemporaryDirectory(prefix="proxy-service-update-") as temporary:
            if xray is None:
                xray = Path(temporary) / "xray"
                install_xray(xray, download_file)
            options = vars(args).copy()
            options.update(config_dir=str(plan.context.config_dir), xray=str(xray),
                           profile="default", enable=True, start=False, restart=True)
            result = service_install(argparse.Namespace(**options))
        if result == 0:
            print("Служба обновлена и готова; её настройки и домашний профиль сохранены.")
        return result
    config = service_config(plan.config_text)
    check_setup_source(plan)
    config_path = plan.context.config_dir / "config"
    current = config_path.read_bytes() if config_path.exists() else None
    if config_path.is_symlink() or current not in (
            plan.original_config, render_machine_config(config, "external").encode()):
        raise ProxyError("домашний config изменился; переход остановлен без его замены. " +
                         setup_reset_hint(plan.context))
    replacing_installed = service_is_installed(Path("/"))
    if not replacing_installed and (plan.manual_state is None or (config.address, config.port) != (
            plan.manual_state["address"], plan.manual_state["port"])):
        check_port_available(ipaddress.ip_address(config.address), config.port)
    with tempfile.TemporaryDirectory(prefix="proxy-service-") as temporary:
        source = Path(temporary)
        install_file(source / "config", plan.config_text.encode(), 0o600)
        install_file(source / "default.url", (plan.profile_text + "\n").encode(), 0o600)
        if xray is None:
            print("Скачивание проверенного Xray {}…".format(XRAY_VERSION))
            xray = source / "xray"
            install_xray(xray, download_file)
        check_xray_version(str(xray))
        generated = source / "client.json"
        write_private_json(generated, build_xray_config(parse_vless(plan.profile_text),
                                                        config.address, config.port))
        run_xray_test(str(xray), generated)
        private_directory(plan.context.state_dir)
        pending = plan.context.state_dir / "service-setup.json"
        install_file(pending, json.dumps({
            "original_config": plan.original_config.decode() if plan.original_config is not None else None,
            "config_text": plan.config_text, "profile_text": plan.profile_text,
            "recovery_file": plan.recovery_file,
            "source_profile": plan.source_profile,
        }).encode(), 0o600)
        options = vars(args).copy()
        options.update(config_dir=str(source), xray=str(xray), profile="default",
                       enable=False, start=False, restart=False, expect_profile=True)
        result = service_install(argparse.Namespace(**options))
        if result:
            if pending.exists():
                sys.stderr.write("Снимок сохранён для повторной попытки. " +
                                 setup_reset_hint(plan.context) + "\n")
            return result
    check_setup_source(plan)
    if managed_xray_running(plan.context):
        if read_state(plan.context) != plan.manual_state:
            raise ProxyError("ручной Xray изменился во время установки; переход остановлен")
        command_stop(plan.context)
    try:
        if replacing_installed:
            service_systemctl("stop", "proxy-client.service")
        check_port_available(ipaddress.ip_address(config.address), config.port)
        service_systemctl("enable", "--now", "proxy-client.service")
        if not wait_until(lambda: service_ready(config), 10):
            raise ProxyError("служба не стала готова")
    except (ProxyError, OSError, subprocess.CalledProcessError, KeyboardInterrupt) as error:
        cleanup = service_systemctl("disable", "--now", "proxy-client.service", check=False)
        recovery = ["proxy-client"]
        if args.profile != "default":
            recovery.extend(["--profile", args.profile])
        recovery.append("start")
        if plan.recovery_file:
            recovery.extend(["-f", plan.recovery_file])
        hint = (" Ручной Xray остановлен; вернуть его: {}.".format(
            " ".join(shlex.quote(part) for part in recovery)) if plan.manual_state else "")
        if cleanup.returncode:
            hint += " Не удалось отключить службу; выполните sudo systemctl disable --now proxy-client."
        raise ProxyError("переход не завершён; домашний config не изменён. "
                         "Проверьте sudo journalctl -u proxy-client -b." + hint) from error
    try:
        switch_home_to_service(plan)
    except (ProxyError, OSError) as error:
        raise ProxyError("служба уже запущена, но домашний профиль не удалось переключить: {}".format(error))
    pending.unlink()
    print("Служба готова на {}:{}. Домашний профиль переключён на external.".format(
        config.address, config.port))
    return 0


def setup_service(args):
    if os.geteuid() == 0:
        raise ProxyError("для настройки и перехода запустите proxy-client service install без sudo; "
                         "для установки только файлов используйте --install-only")
    plan = prepare_service_setup(args)
    xray = find_service_xray(args.xray, Path.home(), Path("/"))
    if xray is None:
        print("Xray {} будет скачан и проверен перед установкой.".format(XRAY_VERSION))
    if plan.config_text is None:
        print("План: обновить и перезапустить службу, включить автозапуск; настройки сохранить.")
    else:
        config = service_config(plan.config_text)
        print("План: установить системную службу на {}:{}, включить автозапуск "
              "и переключить текущий профиль.".format(config.address, config.port))
        if plan.original_config and parse_config(plan.original_config.decode()).proxy_type == "external":
            print("Текущий внешний прокси будет заменён локальной службой; прежний config сохранится.")
        if plan.resume:
            print("Продолжение незавершённого перехода; изменённые настройки не будут заменены.")
        if plan.manual_state is not None:
            print("После установки файлов выбранный ручной Xray будет остановлен; "
                  "текущие соединения прервутся.")
    if not args.yes and input("Продолжить? [y/N]: ").strip().lower() not in ("y", "yes", "д", "да"):
        print("Установка отменена.")
        return 0
    return apply_service_setup(args, plan, xray)


def command_service_install(args):
    if args.port is not None and not 1 <= args.port <= 65535:
        raise ProxyError("порт должен быть числом от 1 до 65535")
    if args.yes and (args.root or args.install_only or args.config_dir):
        raise ProxyError("--yes нельзя совмещать с --root, --install-only или --config-dir")
    if (args.profile_file or args.port is not None) and (args.root or args.install_only or args.config_dir):
        raise ProxyError("--file и --port относятся к настройке; их нельзя совмещать "
                         "с --root, --install-only или --config-dir")
    if args.yes or (sys.stdin.isatty() and not args.root and not args.install_only
                    and not args.config_dir):
        return setup_service(args)
    if not (args.root or args.install_only or args.config_dir):
        raise ProxyError("без терминала выберите --yes для полного перехода "
                         "или --install-only для установки только файлов")
    return service_install(args)


def service_install(args):
    if args.config_dir and args.profile != "default":
        raise ProxyError("выберите --profile или --config-dir, не оба сразу")
    root = Path(args.root) if args.root else Path("/")
    if not root.is_absolute():
        raise ProxyError("root должен быть абсолютным путём")
    if args.root and (args.enable or args.start or args.restart):
        raise ProxyError("--root несовместим с --enable, --start и --restart")
    home = Path.home()
    if os.geteuid() == 0 and "SUDO_UID" in os.environ and (not args.config_dir or not args.xray):
        import pwd
        try:
            home = Path(pwd.getpwuid(int(os.environ["SUDO_UID"])).pw_dir)
        except (ValueError, KeyError):
            raise ProxyError("не удалось определить пользователя sudo; укажите --config-dir и --xray")
    source = Path(args.config_dir) if args.config_dir else user_context(args.profile, home=home).config_dir
    xray = find_service_xray(args.xray, home, root)
    if xray is None:
        raise ProxyError("Xray не найден; запустите proxy-client service install в терминале "
                         "для настройки или выполните proxy-client install")
    if not all(path.is_absolute() for path in (source, xray, root)):
        raise ProxyError("config-dir, xray и root должны быть абсолютными путями")
    if not service_is_installed(root):
        validate_service_source(source)
    if not args.root and os.geteuid() != 0:
        # Пути выбираем до sudo: оно сбрасывает HOME, XDG и PATH.
        command = ["sudo", "--", sys.executable, str(Path(__file__).resolve()),
                   "service", "install", "--config-dir", str(source), "--xray", str(xray)]
        if args.enable:
            command.append("--enable")
        if args.start:
            command.append("--start")
        if args.restart:
            command.append("--restart")
        if args.expect_profile:
            command.append("--expect-profile")
        print("Установка системной службы через sudo.", flush=True)
        return exit_code(subprocess.run(command, check=False).returncode)
    settings = root / "etc/proxy-client"
    # При обновлении служба уже имеет собственные настройки вне HOME.
    config_path = settings / "config" if (settings / "config").exists() else source / "config"
    config = service_config(config_path.read_text(encoding="utf-8"))
    profile_path = settings / "default.url" if (settings / "default.url").exists() else source / "default.url"
    profile_context = service_context(argparse.Namespace(config_dir=config_path.parent,
        runtime_dir=root / "run/proxy-client", xray=xray)) if profile_path.parent == settings else user_context()
    profile = load_profile(profile_path, profile_context)
    if args.expect_profile:
        expected_config = validate_service_source(source)
        expected_profile = load_profile(source / "default.url", user_context())
        if vars(config) != vars(expected_config) or build_xray_config(profile, config.address, config.port) != \
                build_xray_config(expected_profile, expected_config.address, expected_config.port):
            raise ProxyError("системные настройки отличаются от снимка перехода; они оставлены без изменений. "
                             "Для их смены отредактируйте /etc/proxy-client отдельно")
    check_xray_version(str(xray))
    with tempfile.TemporaryDirectory(prefix="proxy-service-test-") as temporary:
        generated = Path(temporary) / "client.json"
        write_private_json(generated, build_xray_config(profile, config.address, config.port))
        run_xray_test(str(xray), generated)
    files = {
        "usr/local/lib/proxy-client/proxy-client": (Path(__file__).read_bytes(), 0o755),
        "usr/local/lib/proxy-client/xray": (xray.read_bytes(), 0o755),
        "etc/systemd/system/proxy-client.service": (SERVICE_UNIT.encode(), 0o644),
        "usr/lib/sysusers.d/proxy-client.conf": (
            b'u proxy-client - "Proxy client" /nonexistent /usr/sbin/nologin\n', 0o644),
    }
    manifest = root / "usr/local/lib/proxy-client/installed.json"
    pending = manifest.with_name("installed.pending.json")
    # До любой записи проверяем все назначения и компоненты пути.
    targets = [root / name for name in files] + [
        manifest, pending, settings, settings / "config", settings / "default.url"]
    for target in targets:
        if any(part.is_symlink() for part in [target, *target.parents]):
            raise ProxyError("установка через симлинк запрещена: {}".format(target))
    owned = json.loads(manifest.read_text()) if manifest.exists() else {}
    interrupted = json.loads(pending.read_text()) if pending.exists() else {}
    for name in files:
        target = root / name
        if target.exists() and (
            not target.is_file() or
            hashlib.sha256(target.read_bytes()).hexdigest() not in
            [owned.get(name)] + interrupted.get(name, [])
        ):
            raise ProxyError("чужой или изменённый файл: {}".format(target))
    if not manifest.exists() and not pending.exists() and settings.exists():
        raise ProxyError("чужой каталог настроек: {}".format(settings))
    make_install_directory(manifest.parent)
    manifest.parent.chmod(0o755)
    # До замены сохраняем допустимые старые и новые хеши. После сбоя повтор
    # распознает уже записанную часть, не принимая произвольные чужие файлы.
    planned = {}
    for name, (data, _) in files.items():
        target = root / name
        planned[name] = [hashlib.sha256(data).hexdigest()]
        if target.exists():
            planned[name].append(hashlib.sha256(target.read_bytes()).hexdigest())
    install_file(pending, json.dumps(planned).encode(), 0o644)
    for name, (data, mode) in files.items():
        install_file(root / name, data, mode)
    settings.mkdir(parents=True, exist_ok=True)
    settings.chmod(0o750)
    for name in ("config", "default.url"):
        target = settings / name
        if not target.exists():
            install_file(target, (source / name).read_bytes(), 0o640)
    install_file(manifest, json.dumps({
        name: hashlib.sha256(data).hexdigest() for name, (data, _) in files.items()
    }).encode(), 0o644)
    pending.unlink()
    if not args.root:
        import grp
        subprocess.run(["systemd-sysusers", "/usr/lib/sysusers.d/proxy-client.conf"], check=True)
        gid = grp.getgrnam("proxy-client").gr_gid
        for path in (settings, settings / "config", settings / "default.url"):
            os.chown(str(path), 0, gid)
        subprocess.run(["systemctl", "daemon-reload"], check=True)
        if args.enable:
            subprocess.run(["systemctl", "enable", "proxy-client.service"], check=True)
        if args.start:
            subprocess.run(["systemctl", "start", "proxy-client.service"], check=True)
        if args.restart:
            subprocess.run(["systemctl", "stop", "proxy-client.service"], check=True)
            check_port_available(ipaddress.ip_address(config.address), config.port)
            subprocess.run(["systemctl", "start", "proxy-client.service"], check=True)
            if not wait_until(lambda: service_ready(config), 10):
                raise ProxyError("служба не стала готова; проверьте sudo journalctl -u proxy-client -b")
    print("Файлы службы установлены.")
    report_journal_storage(root)
    return 0


def build_parser():
    parser = argparse.ArgumentParser(
        prog="proxy-client",
        description="Безопасный запуск команд через явно настроенный прокси.",
    )
    parser.add_argument("--profile", default="default", help="имя пользовательского профиля")
    subparsers = parser.add_subparsers(dest="subcommand")
    service = subparsers.add_parser("service", help="системная служба")
    actions = service.add_subparsers(dest="service_action", required=True)
    serve = actions.add_parser("run", help="запуск из systemd")
    serve.add_argument("--config-dir", default="/etc/proxy-client")
    serve.add_argument("--runtime-dir", default="/run/proxy-client")
    serve.add_argument("--xray", default="/usr/local/lib/proxy-client/xray")
    setup = actions.add_parser("install", help="установить системную службу")
    setup.add_argument("--config-dir", help="каталог настроек (по умолчанию текущий профиль)")
    setup.add_argument("--xray", help="готовый Xray (по умолчанию ~/.local/bin/xray или PATH)")
    setup.add_argument("--root", help="изолированный корень без systemctl")
    setup.add_argument("--enable", action="store_true")
    setup.add_argument("--start", action="store_true")
    setup.add_argument("--restart", action="store_true", help="перезапустить и проверить готовность службы")
    setup.add_argument("--yes", "--auto-yes", action="store_true",
                       help="принять план полного перехода без вопроса")
    setup.add_argument("--install-only", action="store_true",
                       help="установить файлы без мастера и переключения домашнего профиля")
    setup.add_argument("-f", "--file", dest="profile_file", help="закрытый файл VLESS-ссылки для настройки")
    setup.add_argument("--port", type=int, help="локальный порт при настройке (по умолчанию 12334)")
    setup.add_argument("--expect-profile", action="store_true", help=argparse.SUPPRESS)
    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 == "service":
            if arguments.service_action == "install":
                return command_service_install(arguments)
            if arguments.profile != "default":
                raise ProxyError("service run не использует пользовательские профили")
            context = service_context(arguments)
            return command_start(load_config(context), None, context)
        context = user_context(arguments.profile)
        if arguments.subcommand == "install":
            return command_install(arguments.replace, context=context)
        if arguments.subcommand == "stop":
            return command_stop(context)
        config = load_config(context)
        if arguments.subcommand == "status":
            return command_status(config, context)
        if arguments.subcommand == "env":
            return command_env(config, context)
        if arguments.subcommand == "run":
            return command_run(config, arguments.arguments, context)
        if arguments.subcommand == "start":
            return command_start(config, arguments.profile_file, context)
    except (EOFError, KeyboardInterrupt):
        sys.stderr.write("Установка или команда прервана пользователем.\n")
        return 130
    except (ProxyError, OSError, ValueError, subprocess.CalledProcessError) as error:
        sys.stderr.write("Ошибка: {}\n".format(error))
        return 1
    return 2


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