#!/usr/bin/env python3 """proxy-client без управляемого Xray: команда, машинный конфиг, разбор ссылки, install.""" import hashlib import json import os import runpy import socket import subprocess import sys import tempfile import unittest import zipfile from pathlib import Path from unittest import mock from urllib.parse import quote REPO_ROOT = Path(__file__).resolve().parents[1] CLI = REPO_ROOT / "proxy-client/.local/bin/proxy-client" MODULE = runpy.run_path(str(CLI)) ProxyError = MODULE["ProxyError"] def closed_port(): with socket.socket() as probe: probe.bind(("127.0.0.1", 0)) return probe.getsockname()[1] class ExternalProxyCliTest(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.config_dir = self.root / "proxy-client" self.environment = dict( os.environ, XDG_CONFIG_HOME=str(self.root), XDG_STATE_HOME=str(self.root / "state"), PYTHONDONTWRITEBYTECODE="1", ) self.listener = socket.socket() self.addCleanup(self.listener.close) self.listener.bind(("127.0.0.1", 0)) self.listener.listen(16) self.port = self.listener.getsockname()[1] def write_config(self, port=None, proxy_type="external", address="127.0.0.1", no_proxy_extra=".example.test"): self.config_dir.mkdir(exist_ok=True) (self.config_dir / "config").write_text( "[proxy]\ntype = {}\nname = Test proxy\naddress = {}\nport = {}\n" "no_proxy_extra = {}\n".format( proxy_type, address, port or self.port, no_proxy_extra ), encoding="utf-8", ) def run_cli(self, *arguments, **environment): return subprocess.run( [str(CLI)] + list(arguments), env=dict(self.environment, **environment), text=True, capture_output=True, timeout=15, check=False, ) def test_no_subcommand_returns_2_without_side_effects(self): result = self.run_cli() self.assertEqual(2, result.returncode) self.assertEqual("", result.stdout) self.assertIn("укажите явную подкоманду", result.stderr) self.assertEqual([], list(self.root.iterdir())) def test_old_python_gets_clear_error(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.assertIn("требует Python 3.8", result.stderr) def test_status_env_and_run_against_reachable_proxy(self): self.write_config(no_proxy_extra=" .example.test , git.example.test ") status = self.run_cli("status") self.assertEqual(0, status.returncode, status.stderr) self.assertIn("Test proxy (external) доступен", status.stdout) env = self.run_cli("env") proxy_url = "http://127.0.0.1:{}".format(self.port) socks_url = "socks5://127.0.0.1:{}".format(self.port) no_proxy = "127.0.0.1,localhost,::1,[::1],.example.test,git.example.test" self.assertEqual(0, env.returncode, env.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, ], env.stdout.splitlines(), ) code = ( "import json, os, sys; print(json.dumps([sys.argv[1:], " "os.environ['http_proxy']])); sys.exit(42)" ) run = self.run_cli("run", "--", sys.executable, "-c", code, "a b", "--", "--flag") self.assertEqual(42, run.returncode, run.stderr) self.assertEqual([["a b", "--", "--flag"], proxy_url], json.loads(run.stdout)) killed = self.run_cli( "run", "--", sys.executable, "-c", "import os, signal; os.kill(os.getpid(), signal.SIGTERM)", ) self.assertEqual(143, killed.returncode, killed.stderr) def test_unreachable_proxy_prints_nothing_and_runs_nothing(self): self.write_config(port=closed_port()) marker = self.root / "child-ran" for command in (("env",), ("run", "--", "touch", str(marker))): with self.subTest(command=command): result = self.run_cli(*command) self.assertEqual(1, result.returncode) self.assertEqual("", result.stdout) self.assertIn("недоступен", result.stderr) self.assertFalse(marker.exists()) def test_xray_type_without_running_xray_fails_without_fallback(self): self.write_config(proxy_type="xray") marker = self.root / "child-ran" for command in (("status",), ("env",), ("run", "--", "touch", str(marker))): with self.subTest(command=command): result = self.run_cli(*command) self.assertEqual(1, result.returncode) self.assertEqual("", result.stdout) self.assertIn("не запущен", result.stderr) self.assertFalse(marker.exists()) def test_start_refuses_external_type(self): self.write_config() result = self.run_cli("start") self.assertEqual(1, result.returncode) self.assertIn("только для type=xray", result.stderr) def test_missing_config_uses_hiddify_default(self): # Доступен ли Hiddify на этой машине, неважно: и отчёт о доступности, # и ошибка называют адрес по умолчанию. result = self.run_cli("status", PROXY_PREFLIGHT_TIMEOUT="0.05") output = result.stdout + result.stderr self.assertIn("Hiddify", output) self.assertIn("hproxy.local:12334", output) def test_broken_config_symlink_is_an_error(self): self.config_dir.mkdir() (self.config_dir / "config").symlink_to(self.root / "missing") result = self.run_cli("status") self.assertEqual(1, result.returncode) self.assertIn("отсутствующий файл", result.stderr) def test_run_usage_errors(self): self.write_config() missing = self.run_cli("run", "--", "definitely-no-such-command-xyz") self.assertEqual(127, missing.returncode) self.assertIn("не найдена в PATH", missing.stderr) for command in (("run",), ("run", "true"), ("run", "--")): with self.subTest(command=command): self.assertIn("run -- COMMAND", self.run_cli(*command).stderr) def test_invalid_preflight_timeout_is_rejected(self): self.write_config() result = self.run_cli("status", PROXY_PREFLIGHT_TIMEOUT="not-a-number") self.assertEqual(1, result.returncode) self.assertIn("положительным числом", result.stderr) class MachineConfigCliTest(unittest.TestCase): """Машинный конфиг проверяем через команду: код выхода и путь в ошибке.""" VALID = ( "[proxy]\ntype = external\nname = Test\naddress = 127.0.0.1\n" "port = 12334\nno_proxy_extra =\n" ) def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.path = self.root / "proxy-client/config" self.path.parent.mkdir() def start_with(self, text): # start с type=external отказывает сразу после чтения конфига: # сеть и Xray не нужны. self.path.write_text(text, encoding="utf-8") return subprocess.run( [str(CLI), "start"], env=dict(os.environ, XDG_CONFIG_HOME=str(self.root), PYTHONDONTWRITEBYTECODE="1"), text=True, capture_output=True, timeout=15, check=False, ) def test_invalid_configs_are_rejected_with_path(self): cases = { "not ini": "not an ini\n", "unknown section": self.VALID + "[other]\nkey = value\n", "unknown key": self.VALID + "extra = value\n", "duplicate key": self.VALID + "port = 12335\n", "missing key": self.VALID.replace("name = Test\n", ""), "bad type": self.VALID.replace("type = external", "type = socks"), "empty name": self.VALID.replace("name = Test", "name = "), "empty address": self.VALID.replace("address = 127.0.0.1", "address ="), "port zero": self.VALID.replace("port = 12334", "port = 0"), "port high": self.VALID.replace("port = 12334", "port = 65536"), "port syntax": self.VALID.replace("port = 12334", "port = +80"), "empty no-proxy item": self.VALID.replace( "no_proxy_extra =", "no_proxy_extra = one.example,,two.example" ), } for address in ("http://proxy", "[::1]", "has space", "proxy/path", "user@proxy", "прокси.example", "fe80::1%eth0"): cases["address " + address] = self.VALID.replace("127.0.0.1", address) for label, text in cases.items(): with self.subTest(label): result = self.start_with(text) self.assertEqual(1, result.returncode) self.assertIn("Ошибка: машинный конфиг {}:".format(self.path), result.stderr) def test_valid_addresses_are_accepted(self): for address in ("127.0.0.1", "2001:db8::1", "proxy-1.example", "trailing.example."): with self.subTest(address): result = self.start_with(self.VALID.replace("127.0.0.1", address)) self.assertIn("только для type=xray", result.stderr) UUID = "11111111-2222-4333-8444-555555555555" ENCRYPTION = "mlkem768x25519plus.native.0rtt.SYNTHETIC_CLIENT_KEY" REALITY = ( "vless://{}@example.invalid:443?type=tcp&security=reality&encryption=none" "&flow=xtls-rprx-vision&sni=front.example.invalid&fp=chrome" "&pbk=SYNTHETIC_PUBLIC_KEY&sid=SYNTHETIC_SHORT_ID&headerType=none" ).format(UUID) XHTTP = ( "vless://{}@example.invalid:443?type=xhttp&security=tls&encryption=none" "&sni=cdn.example.invalid&fp=chrome&host=cdn.example.invalid&path=%2Fp" ).format(UUID) # Новый формат Hiddify: «/» перед «?», VLESS-шифрование и flow поверх XHTTP. HIDDIFY_XHTTP = ( "vless://{}@192.0.2.1:443/?hiddify=1&fp=chrome&sni=cdn.example.invalid" "&security=tls&alpn=h3&type=xhttp&mode=auto&path=%2Fp&host=cdn.example.invalid" "&extra={}&flow=xtls-rprx-vision&encryption={}#xhttp%20vless%20QUIC" ).format(UUID, quote('{"downloadSettings":{"address":"cdn.example.invalid"}}', safe=""), ENCRYPTION) def outbound(url): profile = MODULE["parse_vless"](url) return MODULE["build_xray_config"](profile, "127.0.0.1", 12334)["outbounds"][0] class VlessProfileTest(unittest.TestCase): def test_reality_mapping(self): result = outbound(REALITY + "&host=ignored.example&alpn=h2") self.assertEqual("xtls-rprx-vision", result["settings"]["flow"]) self.assertEqual( { "network": "raw", "security": "reality", "realitySettings": { "serverName": "front.example.invalid", "fingerprint": "chrome", "password": "SYNTHETIC_PUBLIC_KEY", "shortId": "SYNTHETIC_SHORT_ID", }, }, result["streamSettings"], ) def test_xhttp_mapping_with_all_optional_fields(self): pcs = "a" * 64 + "%2C" + "B" * 64 url = XHTTP + "&alpn=http%2F1.1%2Ch2&mode=auto&pcs={}&vcn=a.example%2Cb.example" \ "&extra=%7B%22x%22%3A1%7D&core=xray".format(pcs) stream = outbound(url)["streamSettings"] self.assertEqual( {"host": "cdn.example.invalid", "path": "/p", "mode": "auto", "extra": {"x": 1}}, stream["xhttpSettings"], ) self.assertEqual( { "serverName": "cdn.example.invalid", "fingerprint": "chrome", "alpn": ["http/1.1", "h2"], "pinnedPeerCertSha256": "a" * 64 + "," + "B" * 64, "verifyPeerCertByName": "a.example,b.example", }, stream["tlsSettings"], ) def test_hiddify_xhttp_with_vless_encryption(self): result = outbound(HIDDIFY_XHTTP) self.assertEqual(ENCRYPTION, result["settings"]["encryption"]) self.assertEqual("xtls-rprx-vision", result["settings"]["flow"]) self.assertEqual("192.0.2.1", result["settings"]["address"]) self.assertEqual(["h3"], result["streamSettings"]["tlsSettings"]["alpn"]) self.assertIn("downloadSettings", result["streamSettings"]["xhttpSettings"]["extra"]) def test_alpn_values(self): for value in ("http/1.1", "h2", "h3", "h2,h3"): with self.subTest(value): tls = outbound(XHTTP + "&alpn=" + quote(value, safe=""))["streamSettings"][ "tlsSettings" ] self.assertEqual(value.split(","), tls["alpn"]) def test_legacy_insecure_is_accepted_ignored_and_flagged_when_true(self): cases = ( ("", False), ("&allowInsecure=false", False), ("&allowInsecure=0&insecure=0", False), ("&insecure=1", True), ("&allowInsecure=TRUE&insecure=false", True), ) for base in (XHTTP, REALITY): for fields, flagged in cases: with self.subTest(base=base[-20:], fields=fields): profile = MODULE["parse_vless"](base + fields) self.assertEqual(flagged, profile.insecure) config = json.dumps(MODULE["build_xray_config"](profile, "127.0.0.1", 1)) self.assertNotIn("nsecure", config) def test_rejected_links(self): cases = { "unknown parameter": REALITY + "&mystery=value", "duplicate parameter": REALITY + "&sni=again.example", "headerType http": REALITY.replace("headerType=none", "headerType=http"), "other flow": REALITY.replace("flow=xtls-rprx-vision", "flow=other"), "bad percent-encoding": REALITY.replace("sni=front.example.invalid", "sni=bad%FF"), "non-root path": REALITY.replace(":443?", ":443/path?"), "grpc": REALITY.replace("type=tcp", "type=grpc"), "missing pbk": REALITY.replace("&pbk=SYNTHETIC_PUBLIC_KEY", ""), "xhttp field in reality": REALITY + "&path=%2Fp", "xhttp with reality": XHTTP.replace("security=tls", "security=reality"), "flow over xhttp without encryption": XHTTP + "&flow=xtls-rprx-vision", "unknown encryption": HIDDIFY_XHTTP.replace("mlkem768x25519plus.", "aes-128-gcm."), "unknown alpn": XHTTP + "&alpn=h4", "empty alpn item": XHTTP + "&alpn=h2%2C", "short pcs": XHTTP + "&pcs=abc", "empty vcn item": XHTTP + "&vcn=a.example%2C", "extra not an object": XHTTP + "&extra=%5B1%5D", "extra not json": XHTTP + "&extra=%7Bx", "bad uuid": REALITY.replace(UUID, "not-a-uuid"), "not vless": REALITY.replace("vless://", "vmess://"), } for label, url in cases.items(): with self.subTest(label): with self.assertRaises(ProxyError): MODULE["parse_vless"](url) class XrayInstallTest(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.home = self.root / "home" self.home.mkdir() self.destination = self.home / ".local/bin/xray" self.environment = { "HOME": str(self.home), "XDG_STATE_HOME": str(self.root / "state"), "PATH": str(self.destination.parent), } self.archive = self.make_archive(self.fake_xray("installed")) @staticmethod def fake_xray(label, version="26.3.27"): return "#!/bin/sh\n[ \"$1\" = version ] && echo 'Xray {} ({})'\n".format( version, label ).encode("ascii") def make_archive(self, content, member="xray"): path = self.root / "archive-{}.zip".format(hashlib.sha256(content).hexdigest()[:8]) with zipfile.ZipFile(str(path), "w") as archive: archive.writestr(member, content) return path.read_bytes() def downloader(self, archive=None, digest=None): archive = self.archive if archive is None else archive if digest is None: digest = ( "MD5= {}\nSHA1= {}\nSHA2-256= {}\n".format( "1" * 32, "2" * 40, hashlib.sha256(archive).hexdigest() ) ) calls = [] def download(url, destination): calls.append(url) if url.endswith(".dgst"): destination.write_text(digest, encoding="ascii") else: destination.write_bytes(archive) return download, calls def install(self, replace=False, downloader=None, machine="x86_64", path=None): if downloader is None: downloader = self.downloader()[0] environment = dict(self.environment) if path is not None: environment["PATH"] = path with mock.patch.dict(os.environ, environment), \ mock.patch("platform.machine", return_value=machine), \ mock.patch("sys.stdout") as stdout: MODULE["command_install"](replace, downloader) return "".join(call.args[0] for call in stdout.write.call_args_list) def write_destination(self, content=b"old sentinel", executable=True): self.destination.parent.mkdir(parents=True, exist_ok=True) self.destination.write_bytes(content) self.destination.chmod(0o755 if executable else 0o644) def assert_destination_kept(self, content=b"old sentinel"): self.assertEqual(content, self.destination.read_bytes()) self.assertEqual([], list(self.destination.parent.glob(".xray-install-*"))) def test_fresh_install(self): downloader, calls = self.downloader() output = self.install(downloader=downloader) self.assertIn("установлен", output) self.assertEqual(0o755, self.destination.stat().st_mode & 0o777) self.assertIn(b"installed", self.destination.read_bytes()) self.assertTrue(calls[0].endswith("/v26.3.27/Xray-linux-64.zip")) self.assertEqual([calls[0], calls[0] + ".dgst"], calls) self.assertEqual(["xray"], [path.name for path in self.destination.parent.iterdir()]) def test_architecture_selects_asset(self): downloader, calls = self.downloader() self.install(downloader=downloader, machine="aarch64") self.assertTrue(calls[0].endswith("/Xray-linux-arm64-v8a.zip")) with self.assertRaisesRegex(ProxyError, "не поддерживается"): self.install(replace=True, downloader=self.fail_download, machine="riscv64") @staticmethod def fail_download(_url, _destination): raise AssertionError("сеть не должна использоваться") def test_digest_is_compared_case_insensitively(self): digest = "SHA2-256= {}\n".format(hashlib.sha256(self.archive).hexdigest().upper()) self.install(downloader=self.downloader(digest=digest)[0]) self.assertTrue(self.destination.exists()) def test_failures_keep_old_binary_and_remove_staging(self): cases = { "wrong hash": self.downloader(digest="SHA2-256= " + "0" * 64 + "\n")[0], "no SHA2-256 line": self.downloader(digest="MD5= " + "1" * 32 + "\n")[0], "corrupt zip": self.downloader(archive=b"not a zip")[0], "no xray in zip": self.downloader( archive=self.make_archive(b"data", member="other") )[0], "wrong version": self.downloader( archive=self.make_archive(self.fake_xray("old", "26.3.26")) )[0], } def interrupted(_url, destination): destination.write_bytes(b"partial") raise KeyboardInterrupt for label, downloader in cases.items(): with self.subTest(label): self.write_destination() with self.assertRaises(ProxyError): self.install(replace=True, downloader=downloader) self.assert_destination_kept() with self.assertRaises(KeyboardInterrupt): self.install(replace=True, downloader=interrupted) self.assert_destination_kept() def test_existing_right_version_is_kept_without_network(self): self.write_destination(self.fake_xray("existing")) output = self.install(downloader=self.fail_download) self.assertIn("уже установлен", output) self.assert_destination_kept(self.fake_xray("existing")) def test_other_existing_file_needs_replace(self): for content, executable in ((b"broken", True), (self.fake_xray("x"), False)): with self.subTest(content=content, executable=executable): self.write_destination(content, executable) with self.assertRaisesRegex(ProxyError, "--replace"): self.install(downloader=self.fail_download) self.install(replace=True) self.assertIn(b"installed", self.destination.read_bytes()) def test_replace_does_not_write_through_symlink(self): target = self.root / "target" target.write_bytes(b"target sentinel") self.destination.parent.mkdir(parents=True) self.destination.symlink_to(target) with self.assertRaisesRegex(ProxyError, "--replace"): self.install(downloader=self.fail_download) self.install(replace=True) self.assertFalse(self.destination.is_symlink()) self.assertEqual(b"target sentinel", target.read_bytes()) def test_directory_destination_is_refused(self): self.destination.mkdir(parents=True) for replace in (False, True): with self.subTest(replace=replace): with self.assertRaisesRegex(ProxyError, "каталог"): self.install(replace=replace, downloader=self.fail_download) def test_path_that_misses_installed_xray_is_an_error(self): competitor_dir = self.root / "competitor" competitor_dir.mkdir() (competitor_dir / "xray").write_bytes(self.fake_xray("competitor")) (competitor_dir / "xray").chmod(0o755) for path in (str(self.root), str(competitor_dir) + os.pathsep + str(self.destination.parent)): with self.subTest(path=path): with self.assertRaisesRegex(ProxyError, "PATH"): self.install(path=path) self.assertTrue(self.destination.is_file()) self.destination.unlink() def test_install_under_running_xray_asks_to_restart_start(self): # Роль работающего Xray играет сам тестовый процесс. state_dir = self.root / "state/proxy-client" state_dir.mkdir(parents=True) (state_dir / "state.json").write_text( json.dumps( { "supervisor_pid": os.getpid(), "xray_pid": os.getpid(), "xray_path": os.path.realpath("/proc/self/exe"), "address": "127.0.0.1", "port": 1, } ), encoding="utf-8", ) self.write_destination(self.fake_xray("old", "26.3.26")) output = self.install(replace=True) self.assertIn("перезапустите proxy-client start", output) self.assertIn(b"installed", self.destination.read_bytes()) def test_install_does_not_read_machine_config(self): self.write_destination(self.fake_xray("existing")) config = self.root / "config/proxy-client" config.mkdir(parents=True) (config / "config").write_text("not an ini\n", encoding="ascii") environment = dict(os.environ, XDG_CONFIG_HOME=str(self.root / "config")) environment.update(self.environment) environment["PATH"] += os.pathsep + os.environ["PATH"] result = subprocess.run( [str(CLI), "install"], env=environment, text=True, capture_output=True, check=False ) self.assertEqual(0, result.returncode, result.stderr) self.assertIn("уже установлен", result.stdout) class PortCheckTest(unittest.TestCase): address = MODULE["ipaddress"].ip_address("127.0.0.1") @staticmethod def listener(): # Как у Xray: Go ставит SO_REUSEADDR на слушающий сокет. sock = socket.socket() sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind(("127.0.0.1", 0)) sock.listen() return sock def test_port_is_free_right_after_xray_closed_its_connections(self): listener = self.listener() port = listener.getsockname()[1] client = socket.create_connection(("127.0.0.1", port)) accepted, _peer = listener.accept() # После Ctrl+C Xray первым закрывает соединения, и ядро держит их # «хвосты» на локальном порту ещё около минуты. accepted.close() listener.close() try: MODULE["check_port_available"](self.address, port) finally: client.close() def test_live_listener_is_reported_as_busy(self): with self.listener() as listener: with self.assertRaisesRegex(ProxyError, "занят"): MODULE["check_port_available"](self.address, listener.getsockname()[1]) if __name__ == "__main__": unittest.main()