#!/usr/bin/env python3 import json import os import subprocess import sys import tempfile import time 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", close_delay=0.0): 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, time\n" "expected_connection = json.loads(os.environ['TEST_EXPECTED_CONNECTION'])\n" "class Connection:\n" " def close(self):\n" " time.sleep(" + repr(close_delay) + ")\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_preflight_timeout_bounds_delayed_dns_resolution(self): self.write_config(12334, address="slow-dns.example") setup = ( "import socket, time; " "socket.getaddrinfo = lambda *args, **kwargs: time.sleep(5)" ) started = time.monotonic() result = self.run_cli_with_socket_setup( ("status",), setup, {"PROXY_PREFLIGHT_TIMEOUT": "0.1"} ) elapsed = time.monotonic() - started self.assertEqual(1, result.returncode) self.assertLess(elapsed, 1.0) self.assertIn("таймаут", result.stderr) self.assertNotIn("Traceback", result.stderr) def test_preflight_timer_is_cancelled_before_closing_successful_connection(self): self.write_config(23456) setup = checking_socket_setup(close_delay=0.1) result = self.run_cli_with_socket_setup( ("status",), setup, {"PROXY_PREFLIGHT_TIMEOUT": "0.01"}, expected_connection=("127.0.0.1", 23456, 0.01), ) self.assertEqual(0, result.returncode, result.stderr) self.assertIn("доступен", result.stdout) self.assertNotIn("Traceback", 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_by_status_env_and_run(self): self.write_config(12334) for value in ("", "zero", "0", "-1", "nan", "inf"): for command in (("status",), ("env",), ("run", "--", "true")): with self.subTest(value=value, command=command): result = self.run_cli( *command, environment={"PROXY_PREFLIGHT_TIMEOUT": value} ) self.assertNotEqual(0, result.returncode) self.assertIn("положительным числом", result.stderr) if command == ("env",): self.assertEqual("", result.stdout) def test_extreme_positive_timeouts_fail_without_traceback(self): self.write_config(12334) for value in ("1e-7", "1e308"): with self.subTest(value=value): result = self.run_cli( "status", environment={"PROXY_PREFLIGHT_TIMEOUT": value} ) self.assertNotEqual(0, result.returncode) self.assertEqual("", result.stdout) self.assertIn("Ошибка:", result.stderr) self.assertNotIn("Traceback", 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", "default section": "[DEFAULT]\nfoo = bar\n" + valid, "unknown key": valid + "extra = value\n", "wrong key case": valid.replace("type = external", "Type = external"), "duplicate key": valid + "port = 12335\n", "missing key": valid.replace("name = Test\n", ""), "multiline value": valid.replace("name = Test", "name = Test\n continued"), "nul value": valid.replace("name = Test", "name = Te\0st").encode("utf-8"), "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_only_ip_literals_or_ascii_dns_names(self): valid_addresses = ("127.0.0.1", "2001:db8::1", "proxy-1.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", "proxy.example:12334", "[::1]", "fe80::1%eth0", "has space", "proxy/path", "user@proxy", ".leading.example", "trailing.example.", "bad..example", "-bad.example", "bad-.example", "прокси.example", ) 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()