dotfiles/tests/test_xray_proxy.py

#!/usr/bin/env python3
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]
CLI = REPO_ROOT / "xray-proxy/.local/bin/xray-proxy"


def checking_socket_setup(outcome="success"):
    if outcome == "success":
        outcome_line = "    return Connection()\n"
    elif outcome == "refused":
        outcome_line = "    raise OSError('connection refused')\n"
    else:
        raise ValueError("unknown socket outcome: {}".format(outcome))
    return (
        "import json, os, socket\n"
        "expected_connection = json.loads(os.environ['TEST_EXPECTED_CONNECTION'])\n"
        "class Connection:\n"
        "    def close(self):\n"
        "        pass\n"
        "def checked_create_connection(endpoint, timeout):\n"
        "    actual_connection = [endpoint[0], endpoint[1], timeout]\n"
        "    if actual_connection != expected_connection:\n"
        "        raise AssertionError('socket call mismatch: expected %r, got %r' "
        "% (expected_connection, actual_connection))\n"
        + outcome_line
        + "socket.create_connection = checked_create_connection"
    )


SUCCESSFUL_SOCKET_SETUP = checking_socket_setup()
REFUSED_SOCKET_SETUP = checking_socket_setup("refused")


class XrayProxyCliTest(unittest.TestCase):
    def setUp(self):
        self.temp_dir = tempfile.TemporaryDirectory()
        self.config_home = Path(self.temp_dir.name)
        self.environment = os.environ.copy()
        self.environment["XDG_CONFIG_HOME"] = str(self.config_home)
        self.environment["PYTHONDONTWRITEBYTECODE"] = "1"

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

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

    def write_raw_config(self, content):
        config_dir = self.config_home / "proxy"
        config_dir.mkdir(exist_ok=True)
        path = config_dir / "config"
        if isinstance(content, bytes):
            path.write_bytes(content)
        else:
            path.write_text(content, encoding="utf-8")

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

    def run_cli_with_socket_setup(
        self, arguments, setup, environment=None, expected_connection=None
    ):
        command_environment = self.environment.copy()
        if environment:
            command_environment.update(environment)
        command_environment["TEST_CLI_ARGUMENTS"] = json.dumps(list(arguments))
        if expected_connection is not None:
            command_environment["TEST_EXPECTED_CONNECTION"] = json.dumps(
                list(expected_connection)
            )
        code = (
            "import json, os, runpy, sys\n"
            + setup
            + "\nsys.argv = [sys.argv[1]] + "
            "json.loads(os.environ['TEST_CLI_ARGUMENTS'])\n"
            "runpy.run_path(sys.argv[0], run_name='__main__')"
        )
        return subprocess.run(
            [sys.executable, "-c", code, str(CLI)],
            text=True,
            capture_output=True,
            env=command_environment,
            check=False,
        )

    def test_socket_fixture_rejects_wrong_endpoint_and_timeout(self):
        self.write_config(23456)
        result = self.run_cli_with_socket_setup(
            ("status",),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("wrong.example", 1, 0.5),
        )

        self.assertNotEqual(0, result.returncode)
        self.assertIn("AssertionError: socket call mismatch", result.stderr)

    def test_no_subcommand_has_no_side_effect_and_reports_usage(self):
        result = self.run_cli()
        self.assertEqual(2, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("укажите явную подкоманду", result.stderr)
        self.assertFalse((self.config_home / "proxy").exists())

    def test_old_python_version_gets_clear_error_before_main_program(self):
        code = (
            "import runpy, sys; "
            "sys.version_info = (3, 7, 9); "
            "runpy.run_path(sys.argv[1], run_name='__main__')"
        )
        result = subprocess.run(
            [sys.executable, "-c", code, str(CLI)],
            text=True,
            capture_output=True,
            check=False,
        )
        self.assertEqual(1, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("требует Python 3.8", result.stderr)
        self.assertIn("Python 3.7", result.stderr)

    def test_broken_config_symlink_is_not_treated_as_missing_config(self):
        config_dir = self.config_home / "proxy"
        config_dir.mkdir()
        (config_dir / "config").symlink_to(self.config_home / "missing-config")
        result = self.run_cli("start")
        self.assertNotEqual(0, result.returncode)
        self.assertIn("отсутствующий файл", result.stderr)
        self.assertNotIn("только для type=xray", result.stderr)

    def test_missing_config_keeps_legacy_external_hiddify_endpoint(self):
        result = self.run_cli("status", environment={"PROXY_PREFLIGHT_TIMEOUT": "0.05"})
        output = result.stdout + result.stderr
        self.assertIn("Hiddify", output)
        self.assertIn("hproxy.local:12334", output)

    def test_status_reports_reachable_external_proxy(self):
        self.write_config(23456)
        result = self.run_cli_with_socket_setup(
            ("status",),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("127.0.0.1", 23456, 2.0),
        )
        self.assertEqual(0, result.returncode, result.stderr)
        self.assertIn("Test proxy (external) доступен", result.stdout)
        self.assertEqual("", result.stderr)

    def test_env_prints_complete_proxy_environment_for_external_proxy(self):
        self.write_config(
            23456, no_proxy_extra=" .example.test , git.example.test "
        )
        result = self.run_cli_with_socket_setup(
            ("env",),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("127.0.0.1", 23456, 2.0),
        )

        proxy_url = "http://127.0.0.1:23456"
        socks_url = "socks5://127.0.0.1:23456"
        no_proxy = (
            "127.0.0.1,localhost,::1,[::1],.example.test,git.example.test"
        )
        self.assertEqual(0, result.returncode, result.stderr)
        self.assertEqual("", result.stderr)
        self.assertEqual(
            [
                "HTTP_PROXY=" + proxy_url,
                "HTTPS_PROXY=" + proxy_url,
                "ALL_PROXY=" + socks_url,
                "NO_PROXY=" + no_proxy,
                "http_proxy=" + proxy_url,
                "https_proxy=" + proxy_url,
                "all_proxy=" + socks_url,
                "no_proxy=" + no_proxy,
            ],
            result.stdout.splitlines(),
        )

    def test_env_brackets_ipv6_without_requiring_host_ipv6(self):
        self.write_config(23456, address="2001:db8::1", no_proxy_extra="")
        result = self.run_cli_with_socket_setup(
            ("env",),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("2001:db8::1", 23456, 2.0),
        )

        self.assertEqual(0, result.returncode, result.stderr)
        self.assertIn("HTTP_PROXY=http://[2001:db8::1]:23456\n", result.stdout)
        self.assertIn("ALL_PROXY=socks5://[2001:db8::1]:23456\n", result.stdout)

    def test_scoped_ipv6_is_rejected_when_ipaddress_lacks_scope_support(self):
        self.write_config(23456, address="fe80::1%eth0", no_proxy_extra="")
        setup = (
            "import ipaddress\n"
            "real_ip_address = ipaddress.ip_address\n"
            "def python38_ip_address(value):\n"
            "    if '%' in value:\n"
            "        raise ValueError('scope IDs unsupported')\n"
            "    return real_ip_address(value)\n"
            "ipaddress.ip_address = python38_ip_address\n"
            + SUCCESSFUL_SOCKET_SETUP
        )
        result = self.run_cli_with_socket_setup(
            ("env",),
            setup,
            expected_connection=("fe80::1%eth0", 23456, 2.0),
        )

        self.assertNotEqual(0, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("address", result.stderr)

    def test_unreachable_env_prints_no_partial_output(self):
        self.write_config(23456)
        result = self.run_cli_with_socket_setup(
            ("env",),
            REFUSED_SOCKET_SETUP,
            {"PROXY_PREFLIGHT_TIMEOUT": "0.1"},
            expected_connection=("127.0.0.1", 23456, 0.1),
        )
        self.assertNotEqual(0, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("недоступен", result.stderr)

    def test_unreachable_external_proxy_never_runs_child(self):
        self.write_config(23456)
        marker = self.config_home / "external-child-ran"
        helper = self.config_home / "external-child"
        helper.write_text(
            "#!/usr/bin/env bash\nprintf ran > \"$1\"\n", encoding="ascii"
        )
        helper.chmod(0o755)

        result = self.run_cli_with_socket_setup(
            ("run", "--", str(helper), str(marker)),
            REFUSED_SOCKET_SETUP,
            {"PROXY_PREFLIGHT_TIMEOUT": "0.1"},
            expected_connection=("127.0.0.1", 23456, 0.1),
        )

        self.assertNotEqual(0, result.returncode)
        self.assertIn("недоступен", result.stderr)
        self.assertFalse(marker.exists())

    def test_xray_type_fails_closed_for_all_ticket_commands(self):
        self.write_config(12334, proxy_type="xray")
        marker = self.config_home / "child-ran"
        helper = self.config_home / "child"
        helper.write_text(
            "#!/usr/bin/env bash\nprintf ran > \"$1\"\n", encoding="ascii"
        )
        helper.chmod(0o755)
        commands = [
            ("status",),
            ("env",),
            ("run", "--", str(helper), str(marker)),
            ("start",),
        ]
        for command in commands:
            with self.subTest(command=command):
                result = self.run_cli(*command)
                self.assertNotEqual(0, result.returncode)
                self.assertEqual("", result.stdout)
                self.assertIn("Ошибка:", result.stderr)
                self.assertFalse(marker.exists())

    def test_external_start_fails_without_starting_anything(self):
        self.write_config(12334)
        result = self.run_cli("start")
        self.assertNotEqual(0, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("только для type=xray", result.stderr)

    def test_run_preserves_arguments_exit_code_and_child_only_environment(self):
        self.write_config(23456)
        code = (
            "import json, os, sys; "
            "print(json.dumps({'args': sys.argv[1:], "
            "'http': os.environ['HTTP_PROXY'], "
            "'lower': os.environ['http_proxy']})); sys.exit(42)"
        )
        original_http = os.environ.get("HTTP_PROXY")
        result = self.run_cli_with_socket_setup(
            ("run", "--", sys.executable, "-c", code, "a b", "--", "--flag"),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("127.0.0.1", 23456, 2.0),
        )

        self.assertEqual(42, result.returncode, result.stderr)
        child = json.loads(result.stdout)
        self.assertEqual(["a b", "--", "--flag"], child["args"])
        expected_url = "http://127.0.0.1:23456"
        self.assertEqual(expected_url, child["http"])
        self.assertEqual(expected_url, child["lower"])
        self.assertEqual(original_http, os.environ.get("HTTP_PROXY"))

    def test_missing_run_command_returns_127(self):
        self.write_config(12334)
        result = self.run_cli("run", "--", "definitely-no-such-command-xyz")
        self.assertEqual(127, result.returncode)
        self.assertEqual("", result.stdout)
        self.assertIn("не найдена в PATH", result.stderr)

    def test_run_maps_child_signal_to_shell_exit_code(self):
        self.write_config(23456)
        result = self.run_cli_with_socket_setup(
            (
                "run",
                "--",
                sys.executable,
                "-c",
                "import os, signal; os.kill(os.getpid(), signal.SIGTERM)",
            ),
            SUCCESSFUL_SOCKET_SETUP,
            expected_connection=("127.0.0.1", 23456, 2.0),
        )
        self.assertEqual(143, result.returncode, result.stderr)

    def test_run_requires_explicit_separator_and_command(self):
        self.write_config(12334)
        for command in (("run",), ("run", "true"), ("run", "--")):
            with self.subTest(command=command):
                result = self.run_cli(*command)
                self.assertNotEqual(0, result.returncode)
                self.assertIn("run -- COMMAND", result.stderr)

    def test_invalid_preflight_timeout_is_rejected_clearly(self):
        self.write_config(12334)
        result = self.run_cli(
            "status", environment={"PROXY_PREFLIGHT_TIMEOUT": "not-a-number"}
        )
        self.assertNotEqual(0, result.returncode)
        self.assertIn("положительным числом", result.stderr)

    def test_ini_rejects_structure_key_and_value_grammar_errors(self):
        valid = (
            "[proxy]\n"
            "type = external\n"
            "name = Test\n"
            "address = 127.0.0.1\n"
            "port = 12334\n"
            "no_proxy_extra =\n"
        )
        invalid_configs = {
            "unknown section": valid + "[other]\nkey = value\n",
            "unknown key": valid + "extra = value\n",
            "duplicate key": valid + "port = 12335\n",
            "missing key": valid.replace("name = Test\n", ""),
            "empty name": valid.replace("name = Test", "name =   "),
            "empty address": valid.replace("address = 127.0.0.1", "address =   "),
            "port zero": valid.replace("port = 12334", "port = 0"),
            "port high": valid.replace("port = 12334", "port = 65536"),
            "port syntax": valid.replace("port = 12334", "port = +80"),
            "empty no-proxy item": valid.replace(
                "no_proxy_extra =", "no_proxy_extra = one.example,,two.example"
            ),
        }
        for label, content in invalid_configs.items():
            with self.subTest(label=label):
                self.write_raw_config(content)
                result = self.run_cli("start")
                self.assertNotEqual(0, result.returncode)
                self.assertEqual("", result.stdout)
                self.assertIn("Ошибка:", result.stderr)

    def test_address_accepts_ip_literals_and_simple_ascii_names(self):
        valid_addresses = (
            "127.0.0.1",
            "2001:db8::1",
            "proxy-1.example",
            ".leading.example",
            "trailing.example.",
            "bad..example",
            "-bad.example",
            "bad-.example",
        )
        for address in valid_addresses:
            with self.subTest(address=address):
                self.write_config(12334, address=address)
                result = self.run_cli("start")
                self.assertIn("только для type=xray", result.stderr)

        invalid_addresses = (
            "http://proxy.example",
            "[::1]",
            "has space",
            "proxy/path",
            "user@proxy",
            "прокси.example",
            "fe80::1%eth0",
        )
        for address in invalid_addresses:
            with self.subTest(address=address):
                self.write_config(12334, address=address)
                result = self.run_cli("start")
                self.assertNotEqual(0, result.returncode)
                self.assertNotIn("только для type=xray", result.stderr)
                self.assertIn("address", result.stderr)


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