#!/usr/bin/env python3
import sys

if sys.version_info < (3, 8):
    sys.stderr.write(
        "Ошибка: xray-proxy требует Python 3.8 или новее; обнаружен Python {}.{}.\n".format(
            sys.version_info[0], sys.version_info[1]
        )
    )
    sys.exit(1)

import argparse
import configparser
import hashlib
import http.client
import ipaddress
import json
import math
import os
import re
import shutil
import signal
import socket
import stat
import subprocess
import time
import uuid
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",)
DNS_LABEL = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?$")
VERSION_LINE = re.compile(r"^Xray (25\.12\.8|26\.3\.27)(?: .*)?$")
QUERY_KEYS = {
    "type",
    "security",
    "encryption",
    "flow",
    "sni",
    "fp",
    "pbk",
    "sid",
    "headerType",
    "host",
    "alpn",
    "hiddify",
    "core",
    "path",
    "mode",
    "extra",
    "pcs",
    "vcn",
    "allowInsecure",
    "insecure",
}
ALPN_VALUES = {"http/1.1", "h2", "h3"}
STATE_KEYS = {
    "supervisor_pid",
    "xray_pid",
    "xray_path",
    "config_path",
    "address",
    "port",
    "profile_fingerprint",
}


class ProxyError(Exception):
    pass


class PreflightExpired(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):
        self.server = server
        self.port = port
        self.profile_id = profile_id
        self.query = query

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

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


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


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


def state_home():
    state_root = os.environ.get("XDG_STATE_HOME")
    if state_root:
        return Path(state_root).expanduser() / "xray-proxy"
    return Path.home() / ".local/state/xray-proxy"


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


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


def validate_address(value):
    if not value or not value.isascii():
        raise ProxyError("поле address должно содержать IPv4, IPv6 или ASCII DNS-имя")
    if any(character.isspace() for character in value):
        raise ProxyError("поле address не должно содержать пробелы")
    if any(separator in value for separator in ("%", "[", "]", "/", "\\", "@", "?", "#")):
        raise ProxyError(
            "поле address должно быть адресом без zone ID, скобок и URL-разделителей"
        )

    try:
        return ipaddress.ip_address(value)
    except ValueError:
        pass

    if len(value) > 253:
        raise ProxyError("DNS-имя в поле address слишком длинное")
    labels = value.split(".")
    if not labels or any(not DNS_LABEL.fullmatch(label) for label in labels):
        raise ProxyError(
            "поле address должно быть адресом без схемы, порта, скобок и URL-разделителей"
        )
    return None


def parse_no_proxy_extra(value):
    if value == "":
        return ()
    items = tuple(item.strip() for item in value.split(","))
    if any(not item for item in items):
        raise ProxyError("no_proxy_extra содержит пустой элемент")
    return items


def default_config():
    return ProxyConfig(
        "external",
        DEFAULT_NAME,
        DEFAULT_ADDRESS,
        DEFAULT_PORT,
        DEFAULT_NO_PROXY_EXTRA,
    )


def load_config():
    path = config_path()
    if not path.exists():
        if path.is_symlink():
            raise ProxyError("машинный конфиг ссылается на отсутствующий файл: {}".format(path))
        return default_config()
    if not path.is_file():
        raise ProxyError("машинный конфиг не является обычным файлом: {}".format(path))

    try:
        raw = path.read_bytes()
    except OSError as error:
        raise ProxyError("не удалось прочитать машинный конфиг {}: {}".format(path, error))
    if b"\0" in raw:
        raise ProxyError("машинный конфиг содержит запрещённый NUL")
    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError:
        raise ProxyError("машинный конфиг должен быть в UTF-8")

    parser = configparser.ConfigParser(
        interpolation=None,
        strict=True,
        allow_no_value=False,
        empty_lines_in_values=False,
    )
    parser.optionxform = str
    try:
        parser.read_string(text, source=str(path))
    except configparser.Error as error:
        raise ProxyError("неверный машинный INI: {}".format(error))

    if parser.defaults():
        raise ProxyError("машинный INI не допускает секцию [DEFAULT]")
    if parser.sections() != ["proxy"]:
        raise ProxyError("машинный INI должен содержать только секцию [proxy]")

    section = parser["proxy"]
    actual_keys = set(section.keys())
    if actual_keys != CONFIG_KEYS:
        missing = sorted(CONFIG_KEYS - actual_keys)
        unknown = sorted(actual_keys - CONFIG_KEYS)
        details = []
        if missing:
            details.append("нет ключей: {}".format(", ".join(missing)))
        if unknown:
            details.append("неизвестные ключи: {}".format(", ".join(unknown)))
        raise ProxyError("неверные ключи [proxy] ({})".format("; ".join(details)))

    values = {key: section.get(key, raw=True) for key in CONFIG_KEYS}
    if any("\n" in value or "\r" in value or "\0" in value for value in values.values()):
        raise ProxyError("многострочные значения в машинном INI запрещены")

    proxy_type = values["type"].strip()
    if proxy_type not in ("external", "xray"):
        raise ProxyError("поле type должно быть external или xray")
    name = values["name"].strip()
    address = values["address"].strip()
    if not name:
        raise ProxyError("поле name не должно быть пустым")
    if not address:
        raise ProxyError("поле address не должно быть пустым")
    validate_address(address)

    port_text = values["port"].strip()
    if not re.fullmatch(r"[0-9]+", port_text):
        raise ProxyError("поле port должно быть целым числом от 1 до 65535")
    port = int(port_text)
    if not 1 <= port <= 65535:
        raise ProxyError("поле port должно быть в диапазоне 1–65535")

    no_proxy_extra = parse_no_proxy_extra(values["no_proxy_extra"])
    return ProxyConfig(proxy_type, name, address, port, no_proxy_extra)


def preflight_timeout():
    raw = os.environ.get("PROXY_PREFLIGHT_TIMEOUT", "2")
    try:
        timeout = float(raw)
    except ValueError:
        raise ProxyError("PROXY_PREFLIGHT_TIMEOUT должен быть положительным числом секунд")
    if not math.isfinite(timeout) or timeout <= 0:
        raise ProxyError("PROXY_PREFLIGHT_TIMEOUT должен быть положительным числом секунд")
    return timeout


def check_reachable(config, timeout):
    def expire_preflight(_signal_number, _frame):
        raise PreflightExpired()

    connection = None
    previous_handler = signal.signal(signal.SIGALRM, expire_preflight)
    try:
        signal.setitimer(signal.ITIMER_REAL, timeout)
        connection = socket.create_connection((config.address, config.port), timeout=timeout)
        signal.setitimer(signal.ITIMER_REAL, 0)
    except PreflightExpired:
        raise ProxyError(
            "прокси {} ({}:{}) недоступен: таймаут {} с".format(
                config.name, config.address, config.port, timeout
            )
        )
    except (OverflowError, ValueError) as error:
        raise ProxyError(
            "PROXY_PREFLIGHT_TIMEOUT={} нельзя применить на этой платформе: {}".format(
                timeout, error
            )
        )
    except OSError as error:
        raise ProxyError(
            "прокси {} ({}:{}) недоступен: {}".format(
                config.name, config.address, config.port, error
            )
        )
    finally:
        signal.setitimer(signal.ITIMER_REAL, 0)
        signal.signal(signal.SIGALRM, previous_handler)
        if connection is not None:
            connection.close()


def check_endpoint(address, port, name="управляемый Xray"):
    config = ProxyConfig("xray", name, address, port, ())
    check_reachable(config, preflight_timeout())


def proxy_host_for_url(address):
    try:
        parsed = ipaddress.ip_address(address)
    except ValueError:
        return address
    if parsed.version == 6:
        return "[{}]".format(address)
    return address


def environment_values(config):
    host = proxy_host_for_url(config.address)
    http_url = "http://{}:{}".format(host, config.port)
    socks_url = "socks5://{}:{}".format(host, config.port)
    no_proxy_items = [config.address, "localhost", "127.0.0.1", "::1", "[::1]"]
    no_proxy_items.extend(config.no_proxy_extra)
    no_proxy = ",".join(dict.fromkeys(no_proxy_items))
    return [
        ("HTTP_PROXY", http_url),
        ("HTTPS_PROXY", http_url),
        ("ALL_PROXY", socks_url),
        ("NO_PROXY", no_proxy),
        ("http_proxy", http_url),
        ("https_proxy", http_url),
        ("all_proxy", socks_url),
        ("no_proxy", no_proxy),
    ]


def private_directory(path):
    path.mkdir(parents=True, exist_ok=True, mode=0o700)
    try:
        path.chmod(0o700)
    except OSError as error:
        raise ProxyError("не удалось закрыть служебный каталог: {}".format(error))


def write_private_json(path, value):
    private_directory(path.parent)
    temporary = path.with_name(path.name + ".tmp.{}".format(os.getpid()))
    descriptor = None
    try:
        descriptor = os.open(
            str(temporary), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
        )
        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
            descriptor = None
            json.dump(
                value,
                stream,
                ensure_ascii=False,
                sort_keys=True,
                allow_nan=False,
            )
            stream.write("\n")
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(str(temporary), str(path))
        path.chmod(0o600)
    except (OSError, ValueError) as error:
        raise ProxyError("не удалось записать закрытый служебный файл: {}".format(error))
    finally:
        if descriptor is not None:
            os.close(descriptor)
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass


def read_profile(path):
    try:
        resolved = path.expanduser().resolve(strict=True)
        file_stat = resolved.stat()
    except (OSError, RuntimeError) as error:
        raise ProxyError("не удалось открыть файл профиля: {}".format(error))
    if not stat.S_ISREG(file_stat.st_mode):
        raise ProxyError("файл профиля должен быть обычным файлом")
    if file_stat.st_mode & 0o077:
        raise ProxyError(
            "конечный файл профиля доступен группе или другим пользователям; выполните chmod 600"
        )
    try:
        raw = resolved.read_bytes()
    except OSError as error:
        raise ProxyError("не удалось прочитать файл профиля: {}".format(error))
    if b"\0" in raw:
        raise ProxyError("файл профиля содержит запрещённый NUL")
    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError:
        raise ProxyError("файл профиля должен быть в UTF-8")
    if text.endswith("\n"):
        text = text[:-1]
    if not text or "\n" in text or "\r" in text or text != text.strip():
        raise ProxyError("файл профиля должен содержать ровно одну непустую VLESS-ссылку")
    return text


def parse_query(raw_query):
    if re.search(r"%(?![0-9A-Fa-f]{2})", raw_query):
        raise ProxyError("профиль содержит неверное URL-кодирование")
    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 any(ord(character) < 32 or ord(character) == 127 for character in key + value):
            raise ProxyError("профиль содержит управляющий символ")
        if key in query:
            raise ProxyError("профиль содержит повтор query-параметра")
        if key not in QUERY_KEYS:
            raise ProxyError("профиль содержит неизвестный query-параметр")
        query[key] = value
    return query


def required(query, *keys):
    missing = [key for key in keys if not query.get(key)]
    if missing:
        raise ProxyError("в профиле отсутствуют обязательные параметры")


def parse_boolean_fields(query):
    values = []
    for key in ("allowInsecure", "insecure"):
        if key in query:
            if query[key] not in ("true", "false"):
                raise ProxyError("legacy-поля принимают только true или false")
            values.append(query[key] == "true")
    if len(values) == 2 and values[0] != values[1]:
        raise ProxyError("allowInsecure и insecure должны совпадать")
    return values[0] if values else False


def parse_json_object(value):
    def reject_duplicates(pairs):
        result = {}
        for key, item in pairs:
            if key in result:
                raise ProxyError("extra содержит повтор ключа JSON")
            result[key] = item
        return result

    def reject_constant(_constant):
        raise ProxyError("extra содержит константу вне стандарта JSON")

    try:
        parsed = json.loads(
            value,
            object_pairs_hook=reject_duplicates,
            parse_constant=reject_constant,
        )
    except ProxyError:
        raise
    except (TypeError, ValueError):
        raise ProxyError("extra должен быть корректным JSON-объектом")
    if not isinstance(parsed, dict):
        raise ProxyError("extra должен быть JSON-объектом")
    return parsed


def parse_vless(url, version):
    try:
        parsed = urlsplit(url)
    except ValueError:
        raise ProxyError("неверная VLESS-ссылка")
    if (
        parsed.scheme != "vless"
        or not parsed.netloc
        or parsed.password is not None
        or parsed.path
    ):
        raise ProxyError("ожидалась одна ссылка vless://UUID@SERVER:PORT")
    try:
        profile_id = unquote(parsed.username or "", encoding="utf-8", errors="strict")
        server = parsed.hostname
        port = parsed.port
    except (UnicodeError, ValueError):
        raise ProxyError("неверные UUID, сервер или порт VLESS")
    try:
        parsed_uuid = uuid.UUID(profile_id)
    except ValueError:
        raise ProxyError("профиль содержит неверный UUID")
    if str(parsed_uuid) != profile_id.lower() or not server or port is None:
        raise ProxyError("профиль содержит неверные UUID, сервер или порт")
    validate_address(server)
    query = parse_query(parsed.query)
    required(query, "type", "security", "encryption", "sni", "fp")
    if query["encryption"] != "none":
        raise ProxyError("поддерживается только encryption=none")
    if query.get("flow") not in (None, "xtls-rprx-vision"):
        raise ProxyError("поддерживается только flow=xtls-rprx-vision")
    if query.get("headerType", "none") != "none":
        raise ProxyError("поддерживается только headerType=none")
    if "hiddify" in query and query["hiddify"] != "1":
        raise ProxyError("поддерживается только метаданные hiddify=1")
    if "core" in query and query["core"] != "xray":
        raise ProxyError("поддерживается только метаданные core=xray")
    legacy_insecure = parse_boolean_fields(query)

    if query["type"] == "tcp" and query["security"] == "reality":
        required(query, "flow", "pbk", "sid")
        forbidden = {"path", "mode", "extra", "pcs", "vcn", "core", "allowInsecure", "insecure"}
        if forbidden.intersection(query):
            raise ProxyError("параметры профиля несовместимы с TCP/REALITY")
    elif query["type"] == "xhttp" and query["security"] == "tls":
        required(query, "host", "path")
        if "flow" in query or "pbk" in query or "sid" in query:
            raise ProxyError("параметры профиля несовместимы с XHTTP/TLS")
        if version == "25.12.8" and ("pcs" in query or "vcn" in query):
            raise ProxyError("pcs/vcn несовместимы с Xray 25.12.8")
        if "pcs" in query:
            hashes = list_value(query["pcs"])
            if any(re.fullmatch(r"[0-9A-Fa-f]{64}", item) is None for item in hashes):
                raise ProxyError("pcs должен содержать 32-байтные SHA-256 в hex")
        if "vcn" in query:
            list_value(query["vcn"])
        if "alpn" in query:
            alpn = query["alpn"].split(",")
            if not alpn or any(item not in ALPN_VALUES for item in alpn):
                raise ProxyError("профиль содержит неподдерживаемый ALPN")
        if "extra" in query:
            parse_json_object(query["extra"])
    else:
        raise ProxyError("поддерживаются только TCP/REALITY и XHTTP/TLS")

    profile = VlessProfile(server, port, profile_id, query)
    profile.legacy_insecure = legacy_insecure
    return profile


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


def build_xray_config(profile, local_address, local_port, version):
    query = profile.query
    settings = {
        "address": profile.server,
        "port": profile.port,
        "id": profile.profile_id,
        "encryption": query["encryption"],
    }
    if "flow" in query:
        settings["flow"] = query["flow"]
    if profile.transport == "tcp":
        reality = {
            "serverName": query["sni"],
            "fingerprint": query["fp"],
            "password": query["pbk"],
            "shortId": query["sid"],
        }
        stream = {
            "network": "raw",
            "security": "reality",
            "realitySettings": reality,
        }
    else:
        xhttp = {"host": query["host"], "path": query["path"]}
        if "mode" in query:
            xhttp["mode"] = query["mode"]
        if "extra" in query:
            xhttp["extra"] = parse_json_object(query["extra"])
        tls = {"serverName": query["sni"], "fingerprint": query["fp"]}
        if "alpn" in query:
            tls["alpn"] = list_value(query["alpn"])
        if version == "26.3.27":
            if "pcs" in query:
                tls["pinnedPeerCertSha256"] = query["pcs"]
            if "vcn" in query:
                tls["verifyPeerCertByName"] = query["vcn"]
        elif profile.legacy_insecure:
            tls["allowInsecure"] = True
        stream = {
            "network": "xhttp",
            "security": "tls",
            "xhttpSettings": xhttp,
            "tlsSettings": tls,
        }
    return {
        "log": {"access": "none", "loglevel": "warning"},
        "inbounds": [
            {
                "tag": "local-proxy",
                "listen": local_address,
                "port": local_port,
                "protocol": "socks",
                "settings": {"auth": "noauth", "udp": True},
            }
        ],
        "outbounds": [
            {
                "tag": "proxy",
                "protocol": "vless",
                "settings": settings,
                "streamSettings": stream,
            }
        ],
    }


def resolve_xray():
    found = shutil.which("xray")
    if found is None:
        raise ProxyError("xray не найден в PATH; сначала выполните xray-proxy install")
    try:
        return str(Path(found).resolve(strict=True))
    except OSError as error:
        raise ProxyError("не удалось разрешить путь к xray: {}".format(error))


def xray_version(path):
    try:
        completed = subprocess.run(
            [path, "version"],
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=5,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        raise ProxyError("не удалось безопасно определить версию выбранного xray")
    first_line = completed.stdout.splitlines()[0] if completed.stdout.splitlines() else ""
    match = VERSION_LINE.fullmatch(first_line)
    if completed.returncode != 0 or match is None:
        raise ProxyError("выбранный xray не сообщил точно поддерживаемую версию")
    return match.group(1)


def run_xray_test(path, config_path):
    try:
        completed = subprocess.run(
            [path, "run", "-test", "-c", str(config_path)],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=10,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        raise ProxyError("проверка созданной конфигурации Xray не выполнена")
    if completed.returncode != 0:
        raise ProxyError("созданная конфигурация не прошла xray run -test")


def process_alive(pid):
    try:
        stat_fields = Path("/proc/{}/stat".format(pid)).read_text(encoding="ascii").split()
        if len(stat_fields) > 2 and stat_fields[2] == "Z":
            return False
        os.kill(pid, 0)
        return True
    except (OSError, ValueError):
        return False


def process_cmdline(pid):
    try:
        raw = Path("/proc/{}/cmdline".format(pid)).read_bytes()
    except OSError:
        return []
    return [item.decode("utf-8", "surrogateescape") for item in raw.split(b"\0") if item]


def process_parent(pid):
    try:
        for line in Path("/proc/{}/status".format(pid)).read_text(encoding="ascii").splitlines():
            if line.startswith("PPid:"):
                return int(line.split()[1])
    except (OSError, ValueError):
        pass
    return None


def process_argument_resolves_to(pid, arguments, expected_path):
    try:
        process_cwd = Path("/proc/{}/cwd".format(pid)).resolve(strict=True)
    except (OSError, RuntimeError):
        return False
    for argument in arguments:
        if not argument or "\0" in argument:
            continue
        candidate = Path(argument)
        if not candidate.is_absolute():
            candidate = process_cwd / candidate
        try:
            if candidate.resolve(strict=True) == expected_path:
                return True
        except (OSError, RuntimeError):
            continue
    return False


def process_socket_inodes(pid):
    inodes = set()
    try:
        descriptors = Path("/proc/{}/fd".format(pid)).iterdir()
        for descriptor in descriptors:
            try:
                target = os.readlink(str(descriptor))
            except OSError:
                continue
            match = re.fullmatch(r"socket:\[([0-9]+)\]", target)
            if match is not None:
                inodes.add(match.group(1))
    except OSError:
        return set()
    return inodes


def decode_proc_address(encoded):
    raw = bytes.fromhex(encoded)
    if len(raw) == 4:
        raw = raw[::-1]
    elif len(raw) == 16:
        raw = b"".join(raw[index : index + 4][::-1] for index in range(0, 16, 4))
    else:
        raise ValueError("unexpected proc address size")
    return ipaddress.ip_address(raw)


def process_owns_listener(pid, address, port):
    inodes = process_socket_inodes(pid)
    if not inodes:
        return False
    expected_address = ipaddress.ip_address(address)
    for table_name in ("tcp", "tcp6"):
        table = Path("/proc/{}/net/{}".format(pid, table_name))
        try:
            lines = table.read_text(encoding="ascii").splitlines()[1:]
        except OSError:
            continue
        for line in lines:
            fields = line.split()
            if len(fields) < 10 or fields[3] != "0A" or fields[9] not in inodes:
                continue
            encoded_address, encoded_port = fields[1].split(":", 1)
            try:
                local_address = decode_proc_address(encoded_address)
                local_port = int(encoded_port, 16)
            except (ValueError, TypeError):
                continue
            if local_address == expected_address and local_port == port:
                return True
    return False


def read_managed_state():
    path = state_path()
    try:
        info = path.lstat()
        if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
            raise ProxyError("состояние управляемого Xray не подтверждено")
        if info.st_uid != os.getuid() or info.st_mode & 0o077:
            raise ProxyError("состояние управляемого Xray не подтверждено")
        value = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        raise ProxyError("управляемый Xray не запущен; выполните xray-proxy start")
    except ProxyError:
        raise
    except (OSError, ValueError, TypeError):
        raise ProxyError("состояние управляемого Xray не подтверждено")
    if not isinstance(value, dict) or set(value) != STATE_KEYS:
        raise ProxyError("состояние управляемого Xray не подтверждено")
    if not all(isinstance(value[key], int) and not isinstance(value[key], bool) for key in ("supervisor_pid", "xray_pid", "port")):
        raise ProxyError("состояние управляемого Xray не подтверждено")
    if not all(isinstance(value[key], str) for key in ("xray_path", "config_path", "address", "profile_fingerprint")):
        raise ProxyError("состояние управляемого Xray не подтверждено")
    return value


def verify_managed_state(config=None, require_port=True):
    state = read_managed_state()
    supervisor = state["supervisor_pid"]
    child = state["xray_pid"]
    if supervisor <= 1 or child <= 1 or supervisor == child:
        raise ProxyError("личность управляемого Xray не подтверждена")
    supervisor_args = process_cmdline(supervisor)
    child_args = process_cmdline(child)
    cli = Path(__file__).resolve()
    if (
        not process_alive(supervisor)
        or not process_argument_resolves_to(supervisor, supervisor_args, cli)
        or "start" not in supervisor_args
        or not process_alive(child)
        or state["xray_path"] not in child_args
        or "run" not in child_args
        or "-c" not in child_args
        or state["config_path"] not in child_args
        or process_parent(child) != supervisor
        or not process_owns_listener(child, state["address"], state["port"])
    ):
        raise ProxyError(
            "личность процесса или listener управляемого Xray не подтверждены"
        )
    if config is not None and (
        config.proxy_type != "xray"
        or config.address != state["address"]
        or config.port != state["port"]
    ):
        raise ProxyError("настройки и состояние управляемого Xray не совпадают")
    if require_port:
        check_endpoint(state["address"], state["port"])
    return state


def safe_summary(profile, local_port, xray_path, version):
    query = profile.query
    print("Xray: {} ({})".format(xray_path, version), flush=True)
    print(
        "Профиль: сервер {}:{}, транспорт {}, защита {}, SNI {}, fingerprint {}, локальный порт {}".format(
            profile.server,
            profile.port,
            "RAW" if profile.transport == "tcp" else "XHTTP",
            profile.security.upper(),
            query["sni"],
            query["fp"],
            local_port,
        ),
        flush=True,
    )
    if profile.transport == "tcp":
        if "host" in query or "alpn" in query:
            print("TCP/REALITY: Host и ALPN из ссылки не используются", flush=True)
        print("Секреты REALITY: скрыты", flush=True)
    else:
        print(
            "XHTTP: Host {}, длина пути: {}, ALPN {}".format(
                query["host"], len(query["path"]), query.get("alpn", "не задан")
            ),
            flush=True,
        )
    print("UUID и ключи: скрыты", flush=True)


def warn_legacy(profile, version):
    if version == "26.3.27" and profile.legacy_insecure:
        sys.stderr.write(
            "Предупреждение: legacy insecure не переносится; сертификат проверяется обычным способом.\n"
        )
    if version == "25.12.8" and profile.legacy_insecure:
        sys.stderr.write(
            "Предупреждение: Xray 25.12.8 использует allowInsecure=true; проверка сертификата отключена.\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": "xray-proxy/1"},
        )
        response = connection.getresponse()
        address = response.read(256).decode("ascii", "replace").strip()
        if response.status != 200 or not address:
            raise ValueError("empty response")
        parsed_address = ipaddress.ip_address(address)
        print("Внешний IP через прокси: {}".format(parsed_address), flush=True)
    except (OSError, ValueError, http.client.HTTPException):
        sys.stderr.write(
            "Предупреждение: внешний IP через прокси определить не удалось; Xray продолжает работу.\n"
        )
    finally:
        connection.close()


def cleanup_managed_files(supervisor_pid):
    try:
        current = read_managed_state()
        if current["supervisor_pid"] == supervisor_pid:
            state_path().unlink()
    except ProxyError:
        pass
    try:
        generated_config_path().unlink()
    except FileNotFoundError:
        pass


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


def command_env(config):
    if config.proxy_type == "external":
        check_reachable(config, preflight_timeout())
    else:
        verify_managed_state(config)
    lines = ["{}={}".format(key, value) for key, value in environment_values(config)]
    sys.stdout.write("\n".join(lines) + "\n")
    return 0


def command_run(config, remainder):
    if not remainder or remainder[0] != "--" or len(remainder) == 1:
        raise ProxyError("использование: xray-proxy run -- COMMAND ARG...")
    command = remainder[1:]
    if shutil.which(command[0]) is None:
        sys.stderr.write("Ошибка: команда не найдена в PATH: {}\n".format(command[0]))
        return 127

    if config.proxy_type == "external":
        check_reachable(config, preflight_timeout())
    else:
        verify_managed_state(config)
    child_environment = os.environ.copy()
    child_environment.update(dict(environment_values(config)))
    try:
        completed = subprocess.run(command, env=child_environment, check=False)
    except FileNotFoundError:
        sys.stderr.write("Ошибка: команда не найдена в PATH: {}\n".format(command[0]))
        return 127
    except OSError as error:
        sys.stderr.write("Ошибка: не удалось запустить {}: {}\n".format(command[0], error))
        return 126
    if completed.returncode < 0:
        return 128 - completed.returncode
    return completed.returncode


def command_start(config, profile_file):
    if config.proxy_type == "external":
        raise ProxyError("start доступен только для type=xray; измените машинный конфиг")
    try:
        address = ipaddress.ip_address(config.address)
    except ValueError:
        raise ProxyError("для type=xray поле address должно быть loopback IP-адресом")
    if not address.is_loopback:
        raise ProxyError("для type=xray требуется loopback IP-адрес")
    if state_path().exists() or state_path().is_symlink():
        raise ProxyError("файл состояния уже существует; управляемый запуск отменён")

    profile_path = Path(profile_file) if profile_file else xray_config_home() / "default.url"
    raw_profile = read_profile(profile_path)
    executable = resolve_xray()
    version = xray_version(executable)
    profile = parse_vless(raw_profile, version)
    generated = build_xray_config(profile, config.address, config.port, version)
    write_private_json(generated_config_path(), generated)
    try:
        run_xray_test(executable, generated_config_path())
        safe_summary(profile, config.port, executable, version)
        warn_legacy(profile, version)
        stopping = {"signal": None}

        def request_stop(signal_number, _frame):
            if stopping["signal"] is None:
                stopping["signal"] = signal_number

        previous = {}
        for handled in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
            previous[handled] = signal.signal(handled, request_stop)
        child = subprocess.Popen(
            [executable, "run", "-c", str(generated_config_path())]
        )
        state = {
            "supervisor_pid": os.getpid(),
            "xray_pid": child.pid,
            "xray_path": executable,
            "config_path": str(generated_config_path()),
            "address": config.address,
            "port": config.port,
            "profile_fingerprint": hashlib.sha256(raw_profile.encode("utf-8")).hexdigest(),
        }
        write_private_json(state_path(), state)
        deadline = time.monotonic() + max(1.0, preflight_timeout())
        last_error = None
        while time.monotonic() < deadline:
            if child.poll() is not None:
                raise ProxyError("Xray завершился до готовности локального порта")
            try:
                check_endpoint(config.address, config.port)
                if not process_owns_listener(child.pid, config.address, config.port):
                    raise ProxyError(
                        "настроенный listener не принадлежит дочернему Xray"
                    )
                last_error = None
                break
            except ProxyError as error:
                last_error = error
                time.sleep(0.03)
        if last_error is not None:
            raise ProxyError(
                "дочерний Xray не подтвердил владение настроенным listener"
            )
        diagnose_ip(config)
        while child.poll() is None and stopping["signal"] is None:
            time.sleep(0.05)
        if stopping["signal"] is not None and child.poll() is None:
            os.kill(child.pid, stopping["signal"])
            try:
                child.wait(timeout=stop_timeout())
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait()
        return_code = child.wait()
        if stopping["signal"] is not None:
            return 0
        if return_code < 0:
            return 128 - return_code
        return return_code
    finally:
        if "previous" in locals():
            for handled, handler in previous.items():
                signal.signal(handled, handler)
        if "child" in locals() and child.poll() is None:
            child.terminate()
            try:
                child.wait(timeout=stop_timeout())
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait()
        cleanup_managed_files(os.getpid())


def stop_timeout():
    raw = os.environ.get("XRAY_PROXY_STOP_TIMEOUT", "5")
    try:
        value = float(raw)
    except ValueError:
        raise ProxyError("XRAY_PROXY_STOP_TIMEOUT должен быть положительным числом секунд")
    if not math.isfinite(value) or value <= 0:
        raise ProxyError("XRAY_PROXY_STOP_TIMEOUT должен быть положительным числом секунд")
    return value


def command_stop():
    timeout = stop_timeout()
    state = verify_managed_state(require_port=True)
    try:
        os.kill(state["supervisor_pid"], signal.SIGTERM)
    except OSError:
        raise ProxyError("не удалось послать сигнал подтверждённому супервизору")
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if (
            not process_alive(state["supervisor_pid"])
            and not process_alive(state["xray_pid"])
            and not state_path().exists()
        ):
            print("Управляемый Xray остановлен")
            return 0
        time.sleep(0.05)
    raise ProxyError("штатная остановка управляемого Xray превысила таймаут")


def build_parser():
    parser = argparse.ArgumentParser(
        prog="xray-proxy",
        description="Безопасный запуск команд через явно настроенный прокси.",
    )
    subparsers = parser.add_subparsers(dest="subcommand")
    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("xray-proxy: ошибка: укажите явную подкоманду\n")
        return 2

    try:
        if arguments.subcommand == "stop":
            return command_stop()
        config = load_config()
        if arguments.subcommand == "status":
            return command_status(config)
        if arguments.subcommand == "env":
            return command_env(config)
        if arguments.subcommand == "run":
            return command_run(config, arguments.arguments)
        if arguments.subcommand == "start":
            return command_start(config, arguments.profile_file)
    except ProxyError as error:
        sys.stderr.write("Ошибка: {}\n".format(error))
        return 1
    return 2


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