dotfiles/tests/test_proxy_client_managed.py

#!/usr/bin/env python3
"""Команда proxy-client с управляемым Xray: argv → код выхода, вывод, файлы, процессы."""
import json
import os
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from urllib.parse import quote


REPO_ROOT = Path(__file__).resolve().parents[1]
CLI = Path(os.environ.get("PROXY_CLIENT_TEST_CLI", str(REPO_ROOT / "proxy-client/.local/bin/proxy-client")))
UUID = "11111111-2222-4333-8444-555555555555"
PATH_SECRET = "/tenant/PATH_SENTINEL_a1b2"
PROFILE = (
    "vless://{}@cdn.example:8443?type=xhttp&security=tls&encryption=none"
    "&sni=tls.example&fp=firefox&alpn=h2&host=host.example&path={}#Synthetic"
).format(UUID, quote(PATH_SECRET, safe=""))
PBK = "PUBLIC_KEY_SENTINEL_7f8a"
SID = "SHORT_ID_SENTINEL_9b3c"
REALITY_PROFILE = (
    "vless://{}@edge.example:443?type=tcp&security=reality&encryption=none"
    "&flow=xtls-rprx-vision&sni=front.example&fp=chrome&pbk={}&sid={}#Synthetic"
).format(UUID, PBK, SID)
READY = "Управляемый Xray работает на"

# Поддельный Xray — симлинк на интерпретатор Python: тогда /proc/PID/exe
# совпадает с разрешённым путём к xray, как у настоящего бинарника.
# «xray version» и «xray run ...» запускают одноимённые скрипты из рабочего
# каталога теста.
FAKE_VERSION = "print('Xray 26.10.10 (fake)')\n"
FAKE_RUN = r'''import json, os, socket, sys

config_path = sys.argv[-1]
if sys.argv[1] == "-test":
    if os.environ.get("FAKE_TEST_FAIL"):
        # Настоящий Xray может напечатать фрагменты конфига — CLI их глушит.
        print(open(config_path).read())
        sys.exit(1)
    sys.exit(0)

inbound = json.load(open(config_path))["inbounds"][0]
server = socket.socket()
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((inbound["listen"], inbound["port"]))
server.listen(16)
while True:
    client, _ = server.accept()
    request = client.recv(4096)
    if b"ifconfig.me" in request and not os.environ.get("FAKE_IP_FAIL"):
        client.sendall(
            b"HTTP/1.1 200 OK\r\nContent-Length: 13\r\nConnection: close\r\n\r\n"
            b"203.0.113.77\n"
        )
    client.close()
'''


def free_port():
    with socket.socket() as listener:
        listener.bind(("127.0.0.1", 0))
        return listener.getsockname()[1]


def pid_alive(pid):
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    return True


class ManagedXrayTest(unittest.TestCase):
    def setUp(self):
        temporary = tempfile.TemporaryDirectory()
        self.addCleanup(temporary.cleanup)
        self.root = Path(os.path.realpath(temporary.name))
        self.addCleanup(self.kill_strays)
        bin_dir = self.root / "bin"
        bin_dir.mkdir()
        self.xray = bin_dir / "xray"
        self.xray.symlink_to(os.path.realpath(sys.executable))
        (self.root / "version").write_text(FAKE_VERSION, encoding="utf-8")
        (self.root / "run").write_text(FAKE_RUN, encoding="utf-8")
        self.config_dir = self.root / "config/proxy-client"
        self.state_dir = self.root / "state/proxy-client"
        self.environment = dict(
            os.environ,
            HOME=str(self.root / "home"),
            XDG_CONFIG_HOME=str(self.root / "config"),
            XDG_STATE_HOME=str(self.root / "state"),
            PATH=str(bin_dir) + os.pathsep + os.environ["PATH"],
            PYTHONDONTWRITEBYTECODE="1",
        )
        self.port = free_port()
        self.write_machine_config()
        self.profile = self.write_profile(self.config_dir / "default.url")

    def test_profiles_run_independently_and_export_selected_environment(self):
        first = self.start()
        named = self.config_dir / "profiles/second"
        named.mkdir(parents=True)
        second_port = free_port()
        (named / "config").write_text(
            (self.config_dir / "config").read_text().replace(str(self.port), str(second_port)))
        # Переопределение -f должно работать без default.url у второго профиля.
        second = self.start("-f", str(self.profile), cli_prefix=["--profile", "second", "start"])
        self.assertIn("proxy-client --profile second stop", self.start_output)
        status = self.run_cli("--profile", "second", "status")
        self.assertEqual(status.returncode, 0, status.stderr)
        env = self.run_cli("--profile", "second", "env")
        self.assertIn("127.0.0.1:" + str(second_port), env.stdout)
        run = self.run_cli("--profile", "second", "run", "--", sys.executable,
                           "-c", "import os,sys; print(os.environ['HTTP_PROXY']); sys.exit(23)")
        self.assertEqual(run.returncode, 23, run.stderr)
        self.assertIn("127.0.0.1:" + str(second_port), run.stdout)
        collision = self.run_cli("--profile", "second", "start", "-f", str(self.profile))
        self.assertEqual(collision.returncode, 1)
        stopped = self.run_cli("--profile", "second", "stop")
        self.assertEqual(stopped.returncode, 0, stopped.stderr)
        second.communicate(timeout=10)
        self.assertIsNone(first.poll())
        self.assertEqual(self.run_cli("--profile", "default", "status").returncode, 0)
        self.assertEqual(self.run_cli("stop").returncode, 0)
        first.communicate(timeout=10)

    def test_service_run_uses_explicit_paths_and_pinned_binary(self):
        runtime = self.root / "service-runtime"
        process = self.start(cli_prefix=[
            "service", "run", "--config-dir", str(self.config_dir),
            "--runtime-dir", str(runtime), "--xray", str(self.xray),
        ], PATH="/usr/bin:/bin")
        self.assertTrue((runtime / "state.json").exists())
        self.assertFalse(self.state_dir.exists())
        process.terminate()
        process.communicate(timeout=10)
        self.assertEqual(process.returncode, 0)
        self.assertFalse((runtime / "client.json").exists())

    def test_system_profile_permission_hint_preserves_group_read(self):
        self.profile.chmod(0o644)
        result = self.run_cli("service", "run", "--config-dir", str(self.config_dir),
                              "--runtime-dir", str(self.root / "system-runtime"),
                              "--xray", str(self.xray))
        self.assertEqual(1, result.returncode)
        self.assertIn("chmod 640", result.stderr)
        self.assertIn("chown root:proxy-client", result.stderr)

    def write_machine_config(self, address="127.0.0.1"):
        self.config_dir.mkdir(parents=True, exist_ok=True)
        (self.config_dir / "config").write_text(
            "[proxy]\ntype = xray\nname = Managed test\naddress = {}\n"
            "port = {}\nno_proxy_extra =\n".format(address, self.port),
            encoding="utf-8",
        )

    def write_profile(self, path, mode=0o600, text=PROFILE):
        path.write_text(text + "\n", encoding="utf-8")
        path.chmod(mode)
        return path

    def run_cli(self, *arguments, **environment):
        return subprocess.run(
            [str(CLI)] + list(arguments),
            cwd=str(self.root),
            env=dict(self.environment, **environment),
            text=True,
            capture_output=True,
            timeout=15,
            check=False,
        )

    def start(self, *arguments, cli_prefix=None, **environment):
        """Запускает start и ждёт строку готовности; возвращает процесс."""
        process = subprocess.Popen(
            [str(CLI)] + (cli_prefix or ["start"]) + list(arguments),
            cwd=str(self.root),
            env=dict(self.environment, **environment),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        self.addCleanup(self.kill_leftovers, process)
        watchdog = threading.Timer(10, process.kill)
        watchdog.start()
        self.start_output = ""
        try:
            for line in process.stdout:
                self.start_output += line
                if READY in line:
                    return process
        finally:
            watchdog.cancel()
        self.fail("start не дошёл до готовности: " + process.communicate()[1])

    def kill_strays(self):
        # Все процессы теста (CLI и поддельный Xray) работают в self.root.
        for entry in Path("/proc").iterdir():
            try:
                if os.readlink(str(entry / "cwd")) == str(self.root):
                    os.kill(int(entry.name), signal.SIGKILL)
            except (OSError, ValueError):
                pass

    def kill_leftovers(self, process):
        self.kill_strays()
        process.communicate()

    def finish(self, process):
        stdout, stderr = process.communicate(timeout=10)
        return self.start_output + stdout, stderr

    def xray_pid(self):
        return json.loads((self.state_dir / "state.json").read_text(encoding="utf-8"))[
            "xray_pid"
        ]

    def assert_no_state_files(self):
        self.assertFalse((self.state_dir / "state.json").exists())
        self.assertFalse((self.state_dir / "client.json").exists())

    def assert_no_secrets(self, text):
        for secret in (UUID, PATH_SECRET, quote(PATH_SECRET, safe=""), PBK, SID):
            self.assertNotIn(secret, text)

    def test_full_lifecycle(self):
        process = self.start()
        xray_pid = self.xray_pid()
        self.assertEqual(0o700, self.state_dir.stat().st_mode & 0o777)
        for name in ("state.json", "client.json"):
            self.assertEqual(0o600, (self.state_dir / name).stat().st_mode & 0o777)

        status = self.run_cli("status")
        env = self.run_cli("env")
        child = self.run_cli(
            "run", "--", sys.executable, "-c", "import os; print(os.environ['ALL_PROXY'])"
        )
        second = self.run_cli("start")
        self.assertEqual(0, status.returncode, status.stderr)
        self.assertIn("управляемый Xray работает", status.stdout)
        self.assertEqual(0, env.returncode, env.stderr)
        self.assertIn("HTTP_PROXY=http://127.0.0.1:{}\n".format(self.port), env.stdout)
        self.assertEqual("socks5://127.0.0.1:{}\n".format(self.port), child.stdout)
        self.assertEqual(1, second.returncode)
        self.assertIn("уже запущен", second.stderr)

        stopped = self.run_cli("stop")
        stdout, stderr = self.finish(process)
        self.assertEqual(0, stopped.returncode, stopped.stderr)
        self.assertEqual(0, process.returncode, stderr)
        self.assertFalse(pid_alive(xray_pid))
        self.assert_no_state_files()
        self.assertIn("Внешний IP через прокси: 203.0.113.77", stdout)
        self.assertIn("длина пути: {}".format(len(PATH_SECRET)), stdout)
        self.assert_no_secrets(stdout + stderr + status.stdout + env.stdout)

    def test_ctrl_c_stops_xray_and_removes_files(self):
        process = self.start()
        xray_pid = self.xray_pid()
        process.send_signal(signal.SIGINT)
        self.finish(process)
        self.assertEqual(0, process.returncode)
        self.assertFalse(pid_alive(xray_pid))
        self.assert_no_state_files()

    def test_xhttp_export_with_empty_host_and_padding_starts(self):
        extra = {"xPaddingBytes": "100-1000", "downloadSettings": {"address": "cdn.example"}}
        profile = PROFILE.replace("&host=host.example", "&host=")
        base, fragment = profile.split("#", 1)
        profile = base + "&x_padding_bytes=100-1000&extra="
        profile += quote(json.dumps(extra), safe="") + "#" + fragment
        self.write_profile(self.profile, text=profile)
        process = self.start()
        config = json.loads((self.state_dir / "client.json").read_text())
        stream = config["outbounds"][0]["streamSettings"]
        self.assertEqual("", stream["xhttpSettings"]["host"])
        self.assertEqual(extra, stream["xhttpSettings"]["extra"])
        process.send_signal(signal.SIGINT)
        stdout, stderr = self.finish(process)
        self.assertEqual(0, process.returncode, stderr)
        self.assert_no_state_files()
        self.assert_no_secrets(stdout + stderr)

    def test_stop_finishes_xray_left_after_killed_supervisor(self):
        process = self.start()
        xray_pid = self.xray_pid()
        process.kill()
        process.wait()

        status = self.run_cli("status")
        self.assertEqual(1, status.returncode)
        self.assertIn("proxy-client stop", status.stderr)
        restarted = self.run_cli("start")
        self.assertEqual(1, restarted.returncode)
        self.assertIn("proxy-client stop", restarted.stderr)

        stopped = self.run_cli("stop")
        self.assertEqual(0, stopped.returncode, stopped.stderr)
        self.assertFalse(pid_alive(xray_pid))
        self.assert_no_state_files()

    def test_stale_files_after_processes_died_are_removed(self):
        self.state_dir.mkdir(parents=True)
        (self.state_dir / "state.json").write_text(
            json.dumps(
                {
                    "supervisor_pid": 999999998,
                    "xray_pid": 999999999,
                    "xray_path": "/x",
                    "address": "127.0.0.1",
                    "port": self.port,
                }
            ),
            encoding="utf-8",
        )
        (self.state_dir / "client.json").write_text("{}", encoding="utf-8")

        result = self.run_cli("status")
        self.assertEqual(1, result.returncode)
        self.assertIn("не запущен", result.stderr)
        self.assert_no_state_files()

    def test_broken_state_file_is_named_in_error(self):
        self.state_dir.mkdir(parents=True)
        state = self.state_dir / "state.json"
        state.write_text("not json", encoding="utf-8")
        result = self.run_cli("stop")
        self.assertEqual(1, result.returncode)
        self.assertIn("rm {}".format(state), result.stderr)

    def test_busy_port_is_reported(self):
        with socket.socket() as listener:
            listener.bind(("127.0.0.1", self.port))
            listener.listen()
            result = self.run_cli("start")
        self.assertEqual(1, result.returncode)
        self.assertIn("порт {} занят".format(self.port), result.stderr)
        self.assert_no_state_files()

    def test_non_loopback_address_is_refused(self):
        self.write_machine_config(address="192.0.2.10")
        result = self.run_cli("start")
        self.assertEqual(1, result.returncode)
        self.assertIn("loopback", result.stderr)

    def test_open_profile_names_target_file_and_gives_chmod_command(self):
        target = self.write_profile(self.root / "real profile.url", mode=0o640)
        link = self.root / "link.url"
        link.symlink_to(target)
        result = self.run_cli("start", "-f", str(link))
        self.assertEqual(1, result.returncode)
        self.assertIn(
            "файл профиля {} доступен другим пользователям; "
            "выполните: chmod 600 '{}'".format(target, target),
            result.stderr,
        )
        self.assert_no_secrets(result.stdout + result.stderr)

    def test_failed_config_test_leaves_no_files_and_no_secrets(self):
        result = self.run_cli("start", FAKE_TEST_FAIL="1")
        self.assertEqual(1, result.returncode)
        self.assertIn("run -test", result.stderr)
        self.assert_no_state_files()
        self.assert_no_secrets(result.stdout + result.stderr)

    def test_failed_ip_probe_is_only_a_warning(self):
        process = self.start(FAKE_IP_FAIL="1")
        self.assertIsNone(process.poll())
        process.send_signal(signal.SIGINT)
        _stdout, stderr = self.finish(process)
        self.assertEqual(0, process.returncode)
        self.assertIn("Предупреждение: внешний IP", stderr)

    def test_reality_secrets_stay_out_of_start_output(self):
        profile = self.write_profile(self.root / "reality.url", text=REALITY_PROFILE)
        process = self.start("-f", str(profile))
        process.send_signal(signal.SIGINT)
        stdout, stderr = self.finish(process)
        self.assertIn("Секреты REALITY: скрыты", stdout)
        self.assert_no_secrets(stdout + stderr)

    def test_wrong_xray_version_is_refused_before_launch(self):
        (self.root / "version").write_text("print('Xray 26.3.26 (fake)')\n", encoding="utf-8")
        result = self.run_cli("start")
        self.assertEqual(1, result.returncode)
        self.assertIn("не сообщил версию Xray 26.10.10", result.stderr)
        self.assert_no_state_files()

    def test_replaced_xray_file_is_still_recognised_and_stopped(self):
        # Нужен настоящий файл, а не симлинк: только тогда замена файла видна
        # в /proc/PID/exe как «(deleted)». Копия интерпретатора находит свою
        # стандартную библиотеку через ../lib.
        interpreter = Path(os.path.realpath(sys.executable))
        self.xray.unlink()
        shutil.copy2(str(interpreter), str(self.xray))
        (self.root / "lib").symlink_to(interpreter.parent.parent / "lib")
        process = self.start()
        xray_pid = self.xray_pid()

        replacement = self.xray.with_name("xray.new")
        shutil.copy2(str(self.xray), str(replacement))
        os.replace(str(replacement), str(self.xray))
        self.assertTrue(os.readlink("/proc/{}/exe".format(xray_pid)).endswith(" (deleted)"))

        status = self.run_cli("status")
        self.assertEqual(0, status.returncode, status.stderr)
        stopped = self.run_cli("stop")
        self.assertEqual(0, stopped.returncode, stopped.stderr)
        self.finish(process)
        self.assertFalse(pid_alive(xray_pid))
        self.assert_no_state_files()

    def test_changed_machine_port_is_refused_while_xray_runs(self):
        process = self.start()
        managed_port = self.port
        with socket.socket() as other:
            other.bind(("127.0.0.1", 0))
            other.listen()
            self.port = other.getsockname()[1]
            self.write_machine_config()
            marker = self.root / "child-ran"
            for command in (("status",), ("env",), ("run", "--", "touch", str(marker))):
                with self.subTest(command=command):
                    result = self.run_cli(*command)
                    self.assertEqual(1, result.returncode)
                    self.assertEqual("", result.stdout)
                    self.assertIn(str(self.config_dir / "config"), result.stderr)
                    self.assertIn(":{}".format(managed_port), result.stderr)
                    self.assertIn("proxy-client stop", result.stderr)
            self.assertFalse(marker.exists())
        process.send_signal(signal.SIGINT)
        self.finish(process)

    def test_bad_link_error_names_profile_file(self):
        target = self.root / "real profile.url"
        target.write_text("vless://not-a-link\n", encoding="utf-8")
        target.chmod(0o600)
        link = self.root / "link.url"
        link.symlink_to(target)
        result = self.run_cli("start", "-f", str(link))
        self.assertEqual(1, result.returncode)
        self.assertIn("Ошибка: файл профиля {}: ".format(target), result.stderr)


if __name__ == "__main__":
    unittest.main()