dotfiles/tests/test_proxy_client_managed.py

#!/usr/bin/env python3
import fcntl
import json
import os
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import time
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"
PBK = "PUBLIC_KEY_SENTINEL_7f8a"
SID = "SHORT_ID_SENTINEL_9b3c"
PATH_SECRET = "/tenant/PATH_SENTINEL_a1b2"
PCS_A = "a" * 64
PCS_B = "b" * 64


FAKE_XRAY_OPERAND = r'''import json
import os
import re
import signal
import socket
import sys
import threading

log_path = os.environ["FAKE_XRAY_LOG"]
operation = os.path.basename(sys.argv[0])

def record(kind, **values):
    values["kind"] = kind
    values["argv"] = [operation] + sys.argv[1:]
    values["executable"] = os.path.realpath("/proc/self/exe")
    with open(log_path, "a", encoding="utf-8") as stream:
        stream.write(json.dumps(values, sort_keys=True) + "\n")

if operation == "version" and not sys.argv[1:]:
    record("version")
    print(os.environ["FAKE_XRAY_VERSION"])
    raise SystemExit(int(os.environ.get("FAKE_VERSION_EXIT", "0")))

is_test = operation == "run" and len(sys.argv) == 4 and sys.argv[1:3] == ["-test", "-c"]
is_run = operation == "run" and len(sys.argv) == 3 and sys.argv[1] == "-c"
if not (is_test or is_run):
    record("bad-argv")
    raise SystemExit(64)

config_path = sys.argv[-1]

def reject_constant(value):
    raise ValueError("non-JSON constant: " + value)

with open(config_path, encoding="utf-8") as stream:
    config = json.load(stream, parse_constant=reject_constant)

if os.environ["FAKE_XRAY_VERSION"].startswith("Xray 26.3.27"):
    tls = config.get("outbounds", [{}])[0].get("streamSettings", {}).get("tlsSettings", {})
    for field in ("pinnedPeerCertSha256", "verifyPeerCertByName"):
        if field in tls and not isinstance(tls[field], str):
            raise SystemExit(65)
    if "pinnedPeerCertSha256" in tls:
        hashes = tls["pinnedPeerCertSha256"].split(",")
        if not hashes or any(re.fullmatch(r"[0-9a-fA-F]{64}", value) is None for value in hashes):
            raise SystemExit(65)
    if "verifyPeerCertByName" in tls and any(
        not value for value in tls["verifyPeerCertByName"].split(",")
    ):
        raise SystemExit(65)

if is_test:
    record("test", config=config)
    if os.environ.get("FAKE_TEST_FAIL") == "1":
        sys.stderr.write(os.environ.get("FAKE_TEST_SECRET", "test failed"))
        raise SystemExit(1)
    raise SystemExit(0)

record("run", config=config)
inbound = config["inbounds"][0]
server = socket.socket()
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((inbound["listen"], inbound["port"]))
server.listen(16)
server.settimeout(0.1)
stopping = False

def stop(signum, _frame):
    global stopping
    record("signal", signum=signum)
    if signum == signal.SIGTERM and os.environ.get("FAKE_IGNORE_TERM") == "1":
        return
    stopping = True

signal.signal(signal.SIGTERM, stop)
signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGHUP, stop)

def serve(client):
    try:
        first = client.recv(1)
        if first == b"\x05":
            methods = client.recv(2)
            if len(methods) == 2:
                client.recv(methods[1])
            client.sendall(b"\x05\x00")
            request = client.recv(4)
            if len(request) != 4:
                return
            address_type = request[3]
            if address_type == 1:
                client.recv(4)
            elif address_type == 3:
                size = client.recv(1)[0]
                client.recv(size)
            elif address_type == 4:
                client.recv(16)
            client.recv(2)
            client.sendall(b"\x05\x00\x00\x01\x7f\x00\x00\x01\x00\x00")
            if client.recv(4) == b"PING":
                client.sendall(b"PONG")
            return
        request = first + client.recv(8192)
        if b"ifconfig.me" in request:
            record("ip-probe", request=request.decode("latin1"))
            if os.environ.get("FAKE_IP_FAIL") == "1":
                return
            body = b"203.0.113.77\n"
        else:
            record("http-request", request=request.decode("latin1"))
            body = b"HTTP-OK"
        client.sendall(
            b"HTTP/1.1 200 OK\r\nContent-Length: "
            + str(len(body)).encode("ascii")
            + b"\r\nConnection: close\r\n\r\n"
            + body
        )
    finally:
        client.close()

while not stopping:
    try:
        client, _address = server.accept()
    except socket.timeout:
        continue
    threading.Thread(target=serve, args=(client,), daemon=True).start()
server.close()
'''


class ManagedXrayCliTest(unittest.TestCase):
    def setUp(self):
        self.temporary = tempfile.TemporaryDirectory()
        self.root = Path(self.temporary.name)
        self.home = self.root / "home"
        self.config_home = self.root / "config"
        self.state_home = self.root / "state"
        self.bin_dir = self.root / "bin"
        self.home.mkdir()
        self.bin_dir.mkdir()
        self.log = self.root / "xray.log"
        self.xray = self.bin_dir / "xray"
        python_prefix = Path(sys.executable).resolve().parent.parent
        python_lib = python_prefix / "lib"
        if python_lib.is_dir():
            (self.root / "lib").symlink_to(python_lib, target_is_directory=True)
        shutil.copy2(sys.executable, self.xray)
        self.xray.chmod(0o755)
        for operand in ("version", "run"):
            (self.root / operand).write_text(FAKE_XRAY_OPERAND, encoding="utf-8")
        self.environment = os.environ.copy()
        self.environment.update(
            {
                "HOME": str(self.home),
                "XDG_CONFIG_HOME": str(self.config_home),
                "XDG_STATE_HOME": str(self.state_home),
                "PATH": str(self.bin_dir) + os.pathsep + self.environment["PATH"],
                "PYTHONDONTWRITEBYTECODE": "1",
                "FAKE_XRAY_LOG": str(self.log),
                "FAKE_XRAY_VERSION": "Xray 26.3.27 (fake)",
                "PROXY_PREFLIGHT_TIMEOUT": "0.5",
                "PROXY_CLIENT_STOP_TIMEOUT": "2",
            }
        )
        self.port = self.free_port()
        self.write_machine_config("xray")

    def tearDown(self):
        self.temporary.cleanup()

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

    def write_machine_config(self, proxy_type):
        directory = self.config_home / "proxy-client"
        directory.mkdir(parents=True, exist_ok=True)
        (directory / "config").write_text(
            "[proxy]\n"
            "type = {}\n"
            "name = Managed test\n"
            "address = 127.0.0.1\n"
            "port = {}\n"
            "no_proxy_extra = .example.test\n".format(proxy_type, self.port),
            encoding="utf-8",
        )

    def write_profile(self, url, explicit=False):
        if explicit:
            path = self.root / "chosen.url"
        else:
            path = self.config_home / "proxy-client/default.url"
            path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(url + "\n", encoding="utf-8")
        path.chmod(0o600)
        return path

    def tcp_profile(self, extra=""):
        return (
            "vless://{}@edge.example:443?type=tcp&security=reality&"
            "encryption=none&flow=xtls-rprx-vision&sni=front.example&fp=chrome&"
            "pbk={}&sid={}&headerType=none&host=ignored.example&alpn=h2&hiddify=1{}"
            "#Synthetic".format(UUID, PBK, SID, extra)
        )

    def xhttp_profile(self, extra=""):
        opaque = quote('{"downloadSettings":{"address":"127.0.0.1"}}', safe="")
        modern_fields = "&pcs={}%2C{}&vcn=name-a.example%2Cname-b.example".format(
            PCS_A, PCS_B
        )
        return (
            "vless://{}@cdn.example:8443?type=xhttp&security=tls&encryption=none&"
            "sni=tls.example&fp=firefox&alpn=http%2F1.1%2Ch2&host=host.example&"
            "path={}&mode=auto&extra={}{}&hiddify=1&core=xray&headerType=none{}#Hiddify"
        ).format(UUID, quote(PATH_SECRET, safe=""), opaque, modern_fields, extra)

    def run_cli(self, *args, environment=None, timeout=8):
        env = self.environment.copy()
        if environment:
            env.update(environment)
        return subprocess.run(
            [str(CLI)] + list(args),
            cwd=str(self.root),
            text=True,
            capture_output=True,
            env=env,
            timeout=timeout,
            check=False,
        )

    def wait_for_state(self, process):
        state = self.state_home / "proxy-client/state.json"
        deadline = time.monotonic() + 5
        while time.monotonic() < deadline:
            if state.exists():
                status = self.run_cli("status")
                if status.returncode == 0:
                    return state
            if process.poll() is not None:
                stdout, stderr = process.communicate()
                self.fail("start exited early: {} {}".format(stdout, stderr))
            time.sleep(0.03)
        self.fail("managed state did not become ready")

    def log_entries(self):
        if not self.log.exists():
            return []
        return [json.loads(line) for line in self.log.read_text().splitlines()]

    def managed_state_path(self):
        return self.state_home / "proxy-client/state.json"

    def read_state(self):
        return json.loads(self.managed_state_path().read_text(encoding="utf-8"))

    def write_state(self, state):
        self.managed_state_path().write_text(
            json.dumps(state) + "\n", encoding="utf-8"
        )
        self.managed_state_path().chmod(0o600)

    def start(self, profile=None):
        args = [str(CLI), "start"]
        if profile is not None:
            args.extend(["-f", str(profile)])
        process = subprocess.Popen(
            args,
            cwd=str(self.root),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=self.environment,
        )
        self.wait_for_state(process)
        return process

    def stop_and_collect(self, process):
        self.write_machine_config("external")
        stopped = self.run_cli("stop")
        stdout, stderr = process.communicate(timeout=4)
        self.assertEqual(0, stopped.returncode, stopped.stderr)
        self.assertEqual(0, process.returncode, stderr)
        return stdout, stderr

    def assert_no_secrets(self, text, profile):
        for secret in (profile, UUID, PBK, SID, PATH_SECRET):
            self.assertNotIn(secret, text)

    def test_complete_managed_lifecycle_for_both_profile_locations(self):
        profiles = ((self.xhttp_profile(), False), (self.tcp_profile(), True))
        for profile_text, explicit in profiles:
            with self.subTest(explicit=explicit):
                profile = self.write_profile(profile_text, explicit=explicit)
                process = self.start(profile if explicit else None)

                status = self.run_cli("status")
                env = self.run_cli("env")
                child = self.run_cli(
                    "run",
                    "--",
                    sys.executable,
                    "-c",
                    "import os; print(os.environ['HTTP_PROXY'])",
                )
                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:{}".format(self.port), env.stdout)
                self.assertEqual(0, child.returncode, child.stderr)
                version_summary = process.stdout.readline()
                self.assertIn(str(self.xray.resolve()), version_summary)
                self.assertIn("26.3.27", version_summary)

                with socket.create_connection(("127.0.0.1", self.port), timeout=1) as http:
                    http.sendall(b"GET http://example.test/ HTTP/1.1\r\nHost: example.test\r\n\r\n")
                    self.assertIn(b"HTTP-OK", http.recv(4096))
                with socket.create_connection(("127.0.0.1", self.port), timeout=1) as socks:
                    socks.sendall(b"\x05\x01\x00")
                    self.assertEqual(b"\x05\x00", socks.recv(2))
                    socks.sendall(b"\x05\x01\x00\x03\x0cexample.test\x00\x50")
                    self.assertEqual(10, len(socks.recv(10)))
                    socks.sendall(b"PING")
                    self.assertEqual(b"PONG", socks.recv(4))

                stdout, stderr = self.stop_and_collect(process)
                entries = self.log_entries()
                kinds = [entry["kind"] for entry in entries]
                self.assertLess(kinds.index("version"), kinds.index("test"))
                self.assertLess(kinds.index("test"), kinds.index("run"))
                self.assertIn("ip-probe", kinds)
                for entry in entries:
                    self.assertEqual(str(self.xray.resolve()), entry["executable"])
                self.assertFalse((self.state_home / "proxy-client/state.json").exists())
                self.assertFalse((self.state_home / "proxy-client/client.json").exists())
                self.assert_no_secrets(stdout + stderr + status.stdout + env.stdout, profile_text)
                self.log.unlink()
                self.port = self.free_port()
                self.write_machine_config("xray")

    def test_generated_xhttp_config_preserves_fields_and_redacts_path(self):
        profile_text = self.xhttp_profile()
        self.write_profile(profile_text)
        process = self.start()
        stdout, stderr = self.stop_and_collect(process)
        config = next(entry["config"] for entry in self.log_entries() if entry["kind"] == "test")
        outbound = config["outbounds"][0]
        stream = outbound["streamSettings"]
        self.assertEqual("xhttp", stream["network"])
        self.assertEqual(
            {"host": "host.example", "path": PATH_SECRET, "mode": "auto", "extra": {"downloadSettings": {"address": "127.0.0.1"}}},
            stream["xhttpSettings"],
        )
        self.assertEqual(
            PCS_A + "," + PCS_B,
            stream["tlsSettings"]["pinnedPeerCertSha256"],
        )
        self.assertEqual(
            "name-a.example,name-b.example",
            stream["tlsSettings"]["verifyPeerCertByName"],
        )
        self.assertEqual(["http/1.1", "h2"], stream["tlsSettings"]["alpn"])
        self.assertEqual("none", config["log"]["access"])
        self.assertIn("длина пути: {}".format(len(PATH_SECRET)), stdout)
        self.assert_no_secrets(stdout + stderr, profile_text)

    def test_reality_mapping_state_shape_modes_and_child_argv_are_closed(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        process = self.start()
        state_path = self.state_home / "proxy-client/state.json"
        config_path = self.state_home / "proxy-client/client.json"
        lock_path = self.state_home / "proxy-client/lock"
        state = json.loads(state_path.read_text(encoding="utf-8"))
        config = json.loads(config_path.read_text(encoding="utf-8"))
        self.assertEqual(0o600, state_path.stat().st_mode & 0o777)
        self.assertEqual(0o600, config_path.stat().st_mode & 0o777)
        self.assertEqual(0o600, lock_path.stat().st_mode & 0o777)
        self.assertEqual("", lock_path.read_text(encoding="ascii"))
        self.assertEqual(0o700, state_path.parent.stat().st_mode & 0o777)
        self.assertEqual(str(self.xray.resolve()), state["xray_path"])
        self.assertEqual("127.0.0.1", state["address"])
        self.assertEqual(self.port, state["port"])
        self.assertEqual(
            {
                "supervisor_pid",
                "xray_pid",
                "xray_path",
                "config_path",
                "address",
                "port",
                "supervisor_path",
                "supervisor_starttime",
                "xray_starttime",
                "boot_id",
            },
            set(state),
        )
        self.assertEqual(str(CLI.resolve()), state["supervisor_path"])
        self.assertIsInstance(state["supervisor_starttime"], int)
        self.assertIsInstance(state["xray_starttime"], int)
        self.assertEqual(
            Path("/proc/sys/kernel/random/boot_id").read_text(encoding="ascii").strip(),
            state["boot_id"],
        )
        state_text = state_path.read_text(encoding="utf-8")
        self.assert_no_secrets(state_text, profile_text)
        stream = config["outbounds"][0]["streamSettings"]
        inbound = config["inbounds"][0]
        self.assertEqual("127.0.0.1", inbound["listen"])
        self.assertEqual(self.port, inbound["port"])
        self.assertEqual("socks", inbound["protocol"])
        self.assertEqual("raw", stream["network"])
        self.assertEqual("reality", stream["security"])
        self.assertEqual(
            {
                "serverName": "front.example",
                "fingerprint": "chrome",
                "password": PBK,
                "shortId": SID,
            },
            stream["realitySettings"],
        )
        self.stop_and_collect(process)
        for entry in self.log_entries():
            argv_text = "\0".join(entry["argv"])
            self.assert_no_secrets(argv_text, profile_text)

    def test_held_lock_rejects_start_without_changing_managed_files(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        state_dir = self.state_home / "proxy-client"
        state_dir.mkdir(parents=True, mode=0o700)
        state_path = state_dir / "state.json"
        config_path = state_dir / "client.json"
        state_path.write_text("state sentinel\n", encoding="ascii")
        config_path.write_text("config sentinel\n", encoding="ascii")
        lock_path = state_dir / "lock"
        with lock_path.open("w", encoding="ascii") as lock:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            result = self.run_cli("start")

        self.assertNotEqual(0, result.returncode)
        self.assertIn("блокиров", result.stderr)
        self.assertEqual("state sentinel\n", state_path.read_text(encoding="ascii"))
        self.assertEqual("config sentinel\n", config_path.read_text(encoding="ascii"))
        self.assertEqual([], self.log_entries())
        self.assert_no_secrets(
            result.stdout + result.stderr + lock_path.read_text(encoding="ascii"),
            profile_text,
        )

    def test_live_managed_instance_rejects_install_replace_without_changes(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        process = self.start()
        state_before = self.managed_state_path().read_bytes()
        xray_before = self.xray.read_bytes()

        result = self.run_cli("install", "--replace")

        self.assertNotEqual(0, result.returncode)
        self.assertIn("сначала", result.stderr)
        self.assertIn("stop", result.stderr)
        self.assertEqual(state_before, self.managed_state_path().read_bytes())
        self.assertEqual(xray_before, self.xray.read_bytes())
        self.assertFalse((self.home / ".local/bin/xray").exists())
        self.assert_no_secrets(result.stdout + result.stderr, profile_text)
        self.stop_and_collect(process)

    def test_stale_cleanup_waits_for_start_lock_and_preserves_files(self):
        state_dir = self.state_home / "proxy-client"
        state_dir.mkdir(parents=True, mode=0o700)
        state = {
            "supervisor_pid": 999999998,
            "xray_pid": 999999999,
            "xray_path": str(self.xray.resolve()),
            "config_path": str(state_dir / "client.json"),
            "address": "127.0.0.1",
            "port": self.port,
            "supervisor_path": str(CLI.resolve()),
            "supervisor_starttime": 1,
            "xray_starttime": 1,
            "boot_id": Path("/proc/sys/kernel/random/boot_id")
            .read_text(encoding="ascii")
            .strip(),
        }
        self.write_state(state)
        config_path = state_dir / "client.json"
        config_path.write_text("stale sentinel\n", encoding="ascii")
        lock_path = state_dir / "lock"
        with lock_path.open("w", encoding="ascii") as lock:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            status = self.run_cli("status")

        self.assertNotEqual(0, status.returncode)
        self.assertIn("блокиров", status.stderr)
        self.assertTrue(self.managed_state_path().exists())
        self.assertEqual("stale sentinel\n", config_path.read_text(encoding="ascii"))

    def test_reused_pid_starttime_is_not_signalled_and_state_is_preserved(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        process = self.start()
        original = self.read_state()
        changed = dict(original)
        changed["xray_starttime"] += 1
        self.write_state(changed)
        marker = self.root / "reused-pid-child-ran"

        for command in (
            ("status",),
            ("env",),
            ("run", "--", "touch", str(marker)),
            ("stop",),
        ):
            with self.subTest(command=command):
                result = self.run_cli(*command)
                self.assertNotEqual(0, result.returncode)
                self.assertIn("вручную", result.stderr)
                self.assertTrue(self.managed_state_path().exists())
        self.assertFalse(marker.exists())
        self.assertTrue(process.poll() is None)

        self.write_state(original)
        stdout, stderr = self.stop_and_collect(process)
        self.assert_no_secrets(stdout + stderr, profile_text)

    def test_wrong_xray_executable_is_not_signalled_and_state_is_preserved(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        process = self.start()
        original = self.read_state()
        changed = dict(original)
        changed["xray_path"] = str(Path("/bin/false").resolve())
        self.write_state(changed)

        stopped = self.run_cli("stop")
        self.assertNotEqual(0, stopped.returncode)
        self.assertIn("вручную", stopped.stderr)
        self.assertTrue(self.managed_state_path().exists())
        self.assertIsNone(process.poll())

        self.write_state(original)
        stdout, stderr = self.stop_and_collect(process)
        self.assert_no_secrets(stdout + stderr, profile_text)

    def test_boot_id_mismatch_cleans_stale_files_without_signalling_processes(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        process = self.start()
        state = self.read_state()
        state["boot_id"] = "00000000-0000-0000-0000-000000000000"
        self.write_state(state)
        os.kill(state["supervisor_pid"], signal.SIGKILL)
        process.wait(timeout=2)

        status = self.run_cli("status")
        self.assertNotEqual(0, status.returncode)
        self.assertIn("устаревшее", status.stderr)
        self.assertFalse(self.managed_state_path().exists())
        self.assertFalse((self.state_home / "proxy-client/client.json").exists())
        self.assertTrue(Path("/proc/{}".format(state["xray_pid"])).exists())

        os.kill(state["xray_pid"], signal.SIGKILL)
        stdout, stderr = process.communicate(timeout=4)
        self.assert_no_secrets(stdout + stderr, profile_text)

    def test_dead_pids_allow_stale_state_and_config_cleanup(self):
        state_dir = self.state_home / "proxy-client"
        state_dir.mkdir(parents=True, mode=0o700)
        state = {
            "supervisor_pid": 999999998,
            "xray_pid": 999999999,
            "xray_path": str(self.xray.resolve()),
            "config_path": str(state_dir / "client.json"),
            "address": "127.0.0.1",
            "port": self.port,
            "supervisor_path": str(CLI.resolve()),
            "supervisor_starttime": 1,
            "xray_starttime": 1,
            "boot_id": Path("/proc/sys/kernel/random/boot_id")
            .read_text(encoding="ascii")
            .strip(),
        }
        self.write_state(state)
        (state_dir / "client.json").write_text("stale\n", encoding="ascii")

        status = self.run_cli("status")
        self.assertNotEqual(0, status.returncode)
        self.assertIn("устаревшее", status.stderr)
        self.assertFalse(self.managed_state_path().exists())
        self.assertFalse((state_dir / "client.json").exists())

    def test_sigkill_supervisor_stop_terminates_confirmed_orphan_and_cleans(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        process = self.start()
        state = self.read_state()
        run_count = len(
            [entry for entry in self.log_entries() if entry["kind"] == "run"]
        )
        alternate_profile = self.write_profile(self.xhttp_profile(), explicit=True)
        repeated = self.run_cli("start")
        switched = self.run_cli("start", "-f", str(alternate_profile))
        for result in (repeated, switched):
            self.assertNotEqual(0, result.returncode)
            self.assertIn("блокиров", result.stderr)
        self.assertEqual(
            run_count,
            len([entry for entry in self.log_entries() if entry["kind"] == "run"]),
        )

        lock_path = self.state_home / "proxy-client/lock"
        with lock_path.open("r+", encoding="ascii") as lock:
            with self.assertRaises(OSError):
                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)

        os.kill(state["supervisor_pid"], signal.SIGKILL)
        process.wait(timeout=2)
        with lock_path.open("r+", encoding="ascii") as lock:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            fcntl.flock(lock, fcntl.LOCK_UN)

        status = self.run_cli("status")
        env = self.run_cli("env")
        marker = self.root / "orphan-child-ran"
        run = self.run_cli("run", "--", "touch", str(marker))
        for result in (status, env, run):
            self.assertNotEqual(0, result.returncode)
            self.assertIn("сирот", result.stderr)
        self.assertFalse(marker.exists())

        stopped = self.run_cli("stop")
        self.assertEqual(0, stopped.returncode, stopped.stderr)
        self.assertIn("сирот", stopped.stdout)
        self.assertFalse(self.managed_state_path().exists())
        self.assertFalse((self.state_home / "proxy-client/client.json").exists())
        stdout, stderr = process.communicate(timeout=2)
        self.assert_no_secrets(stdout + stderr + stopped.stdout, profile_text)

        self.port = self.free_port()
        self.write_machine_config("xray")
        replacement = self.start()
        replacement_stdout, replacement_stderr = self.stop_and_collect(replacement)
        self.assert_no_secrets(
            replacement_stdout + replacement_stderr, profile_text
        )

    def test_unverifiable_live_child_preserves_state_and_config(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        self.environment["PROXY_CLIENT_STOP_TIMEOUT"] = "0.2"
        process = self.start()
        state = self.read_state()
        self.xray.unlink()

        os.kill(state["supervisor_pid"], signal.SIGTERM)
        process.wait(timeout=3)
        state_exists = self.managed_state_path().exists()
        config_path = self.state_home / "proxy-client/client.json"
        config_exists = config_path.exists()
        child_alive = Path("/proc/{}".format(state["xray_pid"])).exists()
        if child_alive:
            os.kill(state["xray_pid"], signal.SIGKILL)
            deadline = time.monotonic() + 2
            while (
                Path("/proc/{}".format(state["xray_pid"])).exists()
                and time.monotonic() < deadline
            ):
                time.sleep(0.02)
        stdout, stderr = process.communicate(timeout=2)

        self.assertTrue(child_alive)
        self.assertTrue(state_exists)
        self.assertTrue(config_exists)
        self.assertNotEqual(0, process.returncode)
        self.assertIn("вручную", stderr)
        self.assert_no_secrets(stdout + stderr, profile_text)

    def test_orphan_ignoring_sigterm_is_rechecked_killed_and_cleaned(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        self.environment["FAKE_IGNORE_TERM"] = "1"
        process = self.start()
        state = self.read_state()
        os.kill(state["supervisor_pid"], signal.SIGKILL)
        process.wait(timeout=2)

        stopped = self.run_cli(
            "stop", environment={"PROXY_CLIENT_STOP_TIMEOUT": "0.2"}
        )
        self.assertEqual(0, stopped.returncode, stopped.stderr)
        self.assertFalse(self.managed_state_path().exists())
        self.assertFalse((self.state_home / "proxy-client/client.json").exists())
        signals = [
            entry["signum"]
            for entry in self.log_entries()
            if entry["kind"] == "signal"
        ]
        self.assertEqual([signal.SIGTERM], signals)
        stdout, stderr = process.communicate(timeout=2)
        self.assert_no_secrets(stdout + stderr + stopped.stdout, profile_text)

    def test_occupied_port_fails_without_touching_listener_or_leaving_generated_files(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        with socket.socket() as listener:
            listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            listener.bind(("127.0.0.1", self.port))
            listener.listen(1)
            result = self.run_cli("start")
            with socket.create_connection(("127.0.0.1", self.port), timeout=1):
                pass

        self.assertNotEqual(0, result.returncode)
        self.assertIn("порт {}".format(self.port), result.stderr)
        self.assertIn("занят", result.stderr)
        self.assertEqual([], self.log_entries())
        self.assertFalse(self.managed_state_path().exists())
        self.assertFalse((self.state_home / "proxy-client/client.json").exists())
        self.assert_no_secrets(result.stdout + result.stderr, profile_text)

    def test_versions_and_profile_failures_stop_before_test_or_run_without_leaks(self):
        invalid_versions = (
            "Xray 25.12.8",
            "Xray 26.3.270",
            "junk Xray 26.3.27",
            "Xray 26.5.9",
            "Xray",
            "",
        )
        for version_line in invalid_versions:
            with self.subTest(version=version_line):
                profile_text = self.tcp_profile()
                self.write_profile(profile_text)
                result = self.run_cli("start", environment={"FAKE_XRAY_VERSION": version_line})
                self.assertNotEqual(0, result.returncode)
                self.assertEqual(["version"], [entry["kind"] for entry in self.log_entries()])
                self.assert_no_secrets(result.stdout + result.stderr, profile_text)
                self.log.unlink()

        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        result = self.run_cli("start", environment={"FAKE_VERSION_EXIT": "1"})
        self.assertNotEqual(0, result.returncode)
        self.assertEqual(["version"], [entry["kind"] for entry in self.log_entries()])
        self.assert_no_secrets(result.stdout + result.stderr, profile_text)
        self.log.unlink()

        invalid_profiles = (
            self.tcp_profile("&mystery=value"),
            self.tcp_profile("&sni=duplicate.example"),
            self.tcp_profile("&headerType=http"),
            self.tcp_profile("&flow=other"),
            self.tcp_profile("&sni=bad%0Aline"),
            self.tcp_profile().replace("sni=front.example", "sni=bad%FF"),
            self.tcp_profile().replace("@edge.example:443?", "@edge.example:443/path?"),
            self.tcp_profile().replace("type=tcp", "type=grpc"),
            self.tcp_profile().replace("&pbk={}".format(PBK), ""),
            self.xhttp_profile("&allowInsecure=True"),
            self.xhttp_profile("&allowInsecure=true&insecure=false"),
            self.xhttp_profile().replace("security=tls", "security=reality"),
        )
        for profile_text in invalid_profiles:
            with self.subTest(profile=profile_text[-50:]):
                self.write_profile(profile_text)
                result = self.run_cli("start")
                self.assertNotEqual(0, result.returncode)
                self.assertEqual(["version"], [entry["kind"] for entry in self.log_entries()])
                self.assert_no_secrets(result.stdout + result.stderr, profile_text)
                self.log.unlink()

    def test_exact_boolean_version_alpn_and_extra_branches(self):
        accepted_alpn = ("http/1.1", "h2", "h3", "http/1.1,h2", "h2,h3")
        for value in accepted_alpn:
            with self.subTest(alpn=value):
                profile_text = self.xhttp_profile().replace(
                    "alpn=http%2F1.1%2Ch2", "alpn=" + quote(value, safe="")
                )
                self.write_profile(profile_text)
                process = self.start()
                self.stop_and_collect(process)
                config = next(
                    entry["config"]
                    for entry in self.log_entries()
                    if entry["kind"] == "test"
                )
                self.assertEqual(
                    value.split(","),
                    config["outbounds"][0]["streamSettings"]["tlsSettings"]["alpn"],
                )
                self.log.unlink()
                self.port = self.free_port()
                self.write_machine_config("xray")

        cases = (
            ("", False),
            ("&allowInsecure=false", False),
            ("&allowInsecure=false&insecure=false", False),
            ("&insecure=true", True),
            ("&allowInsecure=true&insecure=true", True),
        )
        for fields, warns in cases:
            with self.subTest(fields=fields):
                profile_text = self.xhttp_profile(fields)
                self.write_profile(profile_text)
                process = self.start()
                stdout, stderr = self.stop_and_collect(process)
                config = next(
                    entry["config"]
                    for entry in self.log_entries()
                    if entry["kind"] == "test"
                )
                tls = config["outbounds"][0]["streamSettings"]["tlsSettings"]
                self.assertNotIn("allowInsecure", tls)
                self.assertEqual(warns, "Предупреждение" in stderr)
                self.assert_no_secrets(stdout + stderr, profile_text)
                self.log.unlink()
                self.port = self.free_port()
                self.write_machine_config("xray")

        rejected = (
            self.xhttp_profile("&allowInsecure=TRUE"),
            self.xhttp_profile("&allowInsecure=1"),
            self.xhttp_profile("&allowInsecure=true&insecure=false"),
            self.xhttp_profile().replace(
                quote('{"downloadSettings":{"address":"127.0.0.1"}}', safe=""),
                quote('{"same":1,"same":2}', safe=""),
            ),
            self.xhttp_profile().replace("alpn=http%2F1.1%2Ch2", "alpn=h4"),
            self.xhttp_profile().replace(
                quote('{"downloadSettings":{"address":"127.0.0.1"}}', safe=""),
                quote('{"threshold":NaN}', safe=""),
            ),
            self.xhttp_profile().replace(
                quote('{"downloadSettings":{"address":"127.0.0.1"}}', safe=""),
                quote('{"threshold":Infinity}', safe=""),
            ),
            self.xhttp_profile().replace(
                quote('{"downloadSettings":{"address":"127.0.0.1"}}', safe=""),
                quote('{"threshold":-Infinity}', safe=""),
            ),
        )
        self.environment["FAKE_XRAY_VERSION"] = "Xray 26.3.27 (fake)"
        for profile_text in rejected:
            with self.subTest(rejected=profile_text[-45:]):
                self.write_profile(profile_text)
                result = self.run_cli("start")
                self.assertNotEqual(0, result.returncode)
                self.assertEqual(["version"], [entry["kind"] for entry in self.log_entries()])
                if "threshold" in profile_text:
                    self.assertIn("extra", result.stderr)
                    self.assertNotIn("run -test", result.stderr)
                self.assert_no_secrets(result.stdout + result.stderr, profile_text)
                self.log.unlink()

    def test_fake_run_test_rejects_non_json_constants(self):
        for constant in ("NaN", "Infinity", "-Infinity"):
            with self.subTest(constant=constant):
                config = self.root / "invalid-constant.json"
                config.write_text('{"value":' + constant + '}\n', encoding="utf-8")
                result = subprocess.run(
                    [str(self.xray), "run", "-test", "-c", str(config)],
                    cwd=str(self.root),
                    text=True,
                    capture_output=True,
                    env=self.environment,
                    check=False,
                )
                self.assertNotEqual(0, result.returncode)

    def test_failed_ip_diagnostic_is_proxy_observed_warning_and_non_fatal(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        process = self.start()
        # The fake inherited the flag at launch only; restart with the failure flag.
        self.write_machine_config("external")
        self.run_cli("stop")
        process.communicate(timeout=4)
        self.log.unlink()
        self.port = self.free_port()
        self.write_machine_config("xray")
        process = subprocess.Popen(
            [str(CLI), "start"],
            cwd=str(self.root),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=dict(self.environment, FAKE_IP_FAIL="1"),
        )
        self.wait_for_state(process)
        self.assertIsNone(process.poll())
        stdout, stderr = self.stop_and_collect(process)
        self.assertIn("Предупреждение", stderr)
        self.assertIn("ip-probe", [entry["kind"] for entry in self.log_entries()])
        self.assert_no_secrets(stdout + stderr, profile_text)

    def test_immediate_stop_after_first_ready_status_returns_foreground_start(self):
        for attempt in range(3):
            with self.subTest(attempt=attempt):
                profile_text = self.tcp_profile()
                self.write_profile(profile_text)
                process = self.start()
                stdout, stderr = self.stop_and_collect(process)
                self.assertFalse((self.state_home / "proxy-client/state.json").exists())
                self.assertFalse((self.state_home / "proxy-client/client.json").exists())
                self.assert_no_secrets(stdout + stderr, profile_text)
                self.log.unlink()
                self.port = self.free_port()
                self.write_machine_config("xray")

    def test_profile_permissions_follow_symlink_and_test_failure_does_not_leak(self):
        profile_text = self.xhttp_profile()
        target = self.root / "target.url"
        target.write_text(profile_text + "\n", encoding="utf-8")
        target.chmod(0o640)
        link = self.root / "profile.url"
        link.symlink_to(target)
        result = self.run_cli("start", "-f", str(link))
        self.assertNotEqual(0, result.returncode)
        self.assertIn("chmod 600", result.stderr)
        self.assertEqual([], self.log_entries())
        self.assert_no_secrets(result.stdout + result.stderr, profile_text)

        target.chmod(0o600)
        target.write_text(profile_text + "\n" + self.tcp_profile() + "\n", encoding="utf-8")
        result = self.run_cli("start", "-f", str(link))
        self.assertNotEqual(0, result.returncode)
        self.assertIn("ровно одну", result.stderr)
        self.assertEqual([], self.log_entries())
        self.assert_no_secrets(result.stdout + result.stderr, profile_text)

        target.write_text(profile_text + "\n", encoding="utf-8")
        target.chmod(0o600)
        result = self.run_cli(
            "start",
            "-f",
            str(link),
            environment={"FAKE_TEST_FAIL": "1", "FAKE_TEST_SECRET": PATH_SECRET},
        )
        self.assertNotEqual(0, result.returncode)
        self.assertEqual(["version", "test"], [entry["kind"] for entry in self.log_entries()])
        self.assert_no_secrets(result.stdout + result.stderr, profile_text)
        self.assertFalse((self.state_home / "proxy-client/client.json").exists())

    def test_xray_start_requires_loopback_before_selecting_binary(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        config_path = self.config_home / "proxy-client/config"
        config_path.write_text(
            config_path.read_text(encoding="utf-8").replace(
                "address = 127.0.0.1", "address = 192.0.2.10"
            ),
            encoding="utf-8",
        )
        result = self.run_cli("start")
        self.assertNotEqual(0, result.returncode)
        self.assertIn("loopback", result.stderr)
        self.assertEqual([], self.log_entries())
        self.assert_no_secrets(result.stdout + result.stderr, profile_text)

    def test_foreign_state_and_listener_fail_closed_and_are_not_signalled(self):
        listener = socket.socket()
        listener.bind(("127.0.0.1", self.port))
        listener.listen(1)
        state_dir = self.state_home / "proxy-client"
        state_dir.mkdir(parents=True)
        state = {
            "supervisor_pid": os.getpid(),
            "xray_pid": os.getpid(),
            "xray_path": str(self.xray.resolve()),
            "port": self.port,
            "address": "127.0.0.1",
            "config_path": str(state_dir / "client.json"),
        }
        (state_dir / "state.json").write_text(json.dumps(state), encoding="utf-8")
        (state_dir / "state.json").chmod(0o600)
        marker = self.root / "child-ran"
        for command in (("status",), ("env",), ("run", "--", "touch", str(marker)), ("stop",)):
            with self.subTest(command=command):
                result = self.run_cli(*command)
                self.assertNotEqual(0, result.returncode)
                self.assertIn("не подтвержден", result.stderr)
        self.assertFalse(marker.exists())
        listener.close()

    def test_substituted_machine_port_fails_closed_without_touching_foreign_listener(self):
        profile_text = self.tcp_profile()
        self.write_profile(profile_text)
        process = self.start()
        managed_port = self.port
        foreign_port = self.free_port()
        listener = socket.socket()
        listener.bind(("127.0.0.1", foreign_port))
        listener.listen(1)
        config_path = self.config_home / "proxy-client/config"
        config_path.write_text(
            config_path.read_text(encoding="utf-8").replace(
                "port = {}".format(managed_port), "port = {}".format(foreign_port)
            ),
            encoding="utf-8",
        )
        marker = self.root / "substituted-child-ran"
        for command in (
            ("status",),
            ("env",),
            ("run", "--", "touch", str(marker)),
        ):
            with self.subTest(command=command):
                result = self.run_cli(*command)
                self.assertNotEqual(0, result.returncode)
                self.assertIn("не совпадают", result.stderr)
        self.assertFalse(marker.exists())
        stopped = self.run_cli("stop")
        stdout, stderr = process.communicate(timeout=4)
        self.assertEqual(0, stopped.returncode, stopped.stderr)
        self.assertEqual(0, process.returncode, stderr)
        self.assert_no_secrets(stdout + stderr, profile_text)
        listener.settimeout(0.2)
        with socket.create_connection(("127.0.0.1", foreign_port), timeout=1):
            pass
        listener.close()

    def test_supervisor_forwards_original_signal_to_child(self):
        for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
            with self.subTest(signum=signum):
                profile_text = self.tcp_profile()
                self.write_profile(profile_text)
                process = self.start()
                state = json.loads(
                    (self.state_home / "proxy-client/state.json").read_text(
                        encoding="utf-8"
                    )
                )
                os.kill(state["supervisor_pid"], signum)
                stdout, stderr = process.communicate(timeout=4)
                signals = [
                    entry["signum"]
                    for entry in self.log_entries()
                    if entry["kind"] == "signal"
                ]
                self.assertEqual([signum], signals)
                self.assertEqual(0, process.returncode, stderr)
                self.assert_no_secrets(stdout + stderr, profile_text)
                self.log.unlink()
                self.port = self.free_port()
                self.write_machine_config("xray")

    def test_absolute_relative_and_path_symlink_cli_invocations_are_manageable(self):
        proxy_bin = self.root / "proxy-bin"
        proxy_bin.mkdir()
        (proxy_bin / "proxy-client").symlink_to(CLI)
        invocations = (
            ([str(CLI)], self.root, self.environment),
            ([os.path.relpath(str(CLI), str(self.root))], self.root, self.environment),
            (
                ["proxy-client"],
                self.root,
                dict(
                    self.environment,
                    PATH=str(proxy_bin)
                    + os.pathsep
                    + str(self.bin_dir)
                    + os.pathsep
                    + os.environ["PATH"],
                ),
            ),
        )
        for command, cwd, environment in invocations:
            with self.subTest(command=command):
                profile_text = self.tcp_profile()
                self.write_profile(profile_text)
                process = subprocess.Popen(
                    command + ["start"],
                    cwd=str(cwd),
                    text=True,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    env=environment,
                )
                state_path = self.state_home / "proxy-client/state.json"
                deadline = time.monotonic() + 3
                while time.monotonic() < deadline and not state_path.exists():
                    if process.poll() is not None:
                        break
                    time.sleep(0.02)
                status = self.run_cli("status")
                if status.returncode == 0:
                    stopped = self.run_cli("stop")
                else:
                    stopped = None
                    if process.poll() is None:
                        os.kill(process.pid, signal.SIGTERM)
                stdout, stderr = process.communicate(timeout=4)
                self.assertEqual(0, status.returncode, status.stderr)
                self.assertIsNotNone(stopped)
                self.assertEqual(0, stopped.returncode, stopped.stderr)
                self.assertEqual(0, process.returncode, stderr)
                self.assert_no_secrets(stdout + stderr, profile_text)
                self.log.unlink()
                self.port = self.free_port()
                self.write_machine_config("xray")


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