dotfiles/tests/test_xray_proxy_managed.py

#!/usr/bin/env python3
import json
import os
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 / "xray-proxy/.local/bin/xray-proxy"
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 = r'''#!/usr/bin/env python3
import json
import os
import re
import signal
import socket
import sys
import threading

log_path = os.environ["FAKE_XRAY_LOG"]

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

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

is_test = len(sys.argv) == 5 and sys.argv[1:4] == ["run", "-test", "-c"]
is_run = len(sys.argv) == 4 and sys.argv[1:3] == ["run", "-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 sys.argv[2] == "-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)
    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"
        self.xray.write_text(FAKE_XRAY, encoding="utf-8")
        self.xray.chmod(0o755)
        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",
                "XRAY_PROXY_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"
        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 / "xray-proxy/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),
            text=True,
            capture_output=True,
            env=env,
            timeout=timeout,
            check=False,
        )

    def wait_for_state(self, process):
        state = self.state_home / "xray-proxy/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 start(self, profile=None):
        args = [str(CLI), "start"]
        if profile is not None:
            args.extend(["-f", str(profile)])
        process = subprocess.Popen(
            args,
            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 / "xray-proxy/state.json").exists())
                self.assertFalse((self.state_home / "xray-proxy/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 / "xray-proxy/state.json"
        config_path = self.state_home / "xray-proxy/client.json"
        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(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",
            },
            set(state),
        )
        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_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)],
                    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"],
            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 / "xray-proxy/state.json").exists())
                self.assertFalse((self.state_home / "xray-proxy/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 / "xray-proxy/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/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 / "xray-proxy"
        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/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 / "xray-proxy/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 / "xray-proxy").symlink_to(CLI)
        invocations = (
            ([str(CLI)], REPO_ROOT, self.environment),
            ([os.path.relpath(str(CLI), str(REPO_ROOT))], REPO_ROOT, self.environment),
            (
                ["xray-proxy"],
                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 / "xray-proxy/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()