dotfiles/tests/test_proxy_client_managed.py

#!/usr/bin/env python3
"""Команда proxy-client с управляемым Xray: argv → код выхода, вывод, файлы, процессы."""
import json
import os
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 = 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=""))
READY = "остановка: Ctrl+C или proxy-client stop"

# Поддельный Xray — симлинк на интерпретатор Python: тогда /proc/PID/exe
# совпадает с разрешённым путём к xray, как у настоящего бинарника.
# «xray version» и «xray run ...» запускают одноимённые скрипты из рабочего
# каталога теста.
FAKE_VERSION = "print('Xray 26.3.27 (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(temporary.name)
        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 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):
        path.write_text(PROFILE + "\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, **environment):
        """Запускает start и ждёт строку готовности; возвращает процесс."""
        process = subprocess.Popen(
            [str(CLI), "start"],
            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_leftovers(self, process):
        state = self.state_dir / "state.json"
        if state.exists():
            pid = json.loads(state.read_text(encoding="utf-8"))["xray_pid"]
            if pid_alive(pid):
                os.kill(pid, signal.SIGKILL)
        if process.poll() is None:
            process.kill()
        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="")):
            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_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"}
            ),
            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)


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