#!/usr/bin/env python3 import json import os import signal import socket import subprocess import sys import tempfile import threading 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) listen_port = int(os.environ.get("FAKE_XRAY_LISTEN_PORT", inbound["port"])) server.bind((inbound["listen"], listen_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="", modern=True): 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 ) if modern else "" ) 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, version="26.3.27"): self.environment["FAKE_XRAY_VERSION"] = "Xray {} (fake)".format(version) 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_versions_and_profile_locations(self): profiles = (("26.3.27", self.xhttp_profile(), False), ("25.12.8", self.tcp_profile(), True)) for version, profile_text, explicit in profiles: with self.subTest(version=version): profile = self.write_profile(profile_text, explicit=explicit) process = self.start(profile if explicit else None, version) 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) self.assertIn(str(self.xray.resolve()), process.stdout.readline()) 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(64, len(state["profile_fingerprint"])) 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_old_version_preserves_explicit_legacy_insecure(self): profile_text = self.xhttp_profile( "&allowInsecure=true&insecure=true", modern=False ) self.write_profile(profile_text) process = self.start(version="25.12.8") 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] self.assertEqual("xhttp", outbound["streamSettings"]["network"]) self.assertTrue(outbound["streamSettings"]["tlsSettings"]["allowInsecure"]) self.assertIn("Предупреждение", stderr) self.assert_no_secrets(stdout + stderr, profile_text) def test_versions_and_profile_failures_stop_before_test_or_run_without_leaks(self): invalid_versions = ( "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("@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 = ( ("26.3.27", "", False, False), ("26.3.27", "&allowInsecure=false", False, False), ("26.3.27", "&allowInsecure=false&insecure=false", False, False), ("26.3.27", "&insecure=true", True, False), ("26.3.27", "&allowInsecure=true&insecure=true", True, False), ("25.12.8", "", False, False), ("25.12.8", "&insecure=false", False, False), ("25.12.8", "&allowInsecure=false&insecure=false", False, False), ("25.12.8", "&allowInsecure=true", True, True), ("25.12.8", "&allowInsecure=true&insecure=true", True, True), ) for version, fields, warns, emitted in cases: with self.subTest(version=version, fields=fields): profile_text = self.xhttp_profile(fields, modern=version == "26.3.27") self.write_profile(profile_text) process = self.start(version=version) 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.assertEqual(emitted, tls.get("allowInsecure", False)) 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() old_with_modern = self.xhttp_profile() self.write_profile(old_with_modern) result = self.run_cli( "start", environment={"FAKE_XRAY_VERSION": "Xray 25.12.8 (fake)"} ) self.assertNotEqual(0, result.returncode) self.assertIn("pcs/vcn", result.stderr) self.assertEqual(["version"], [entry["kind"] for entry in self.log_entries()]) 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"), "profile_fingerprint": "0" * 64, } (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_verified_child_must_own_configured_listener(self): profile_text = self.tcp_profile() self.write_profile(profile_text) wrong_port = self.free_port() listener = socket.socket() listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) listener.bind(("127.0.0.1", self.port)) listener.listen(16) listener.settimeout(0.05) serving = threading.Event() serving.set() def serve_foreign(): while serving.is_set(): try: client, _address = listener.accept() except socket.timeout: continue try: client.settimeout(0.2) request = client.recv(8192) if b"ifconfig.me" in request: body = b"203.0.113.88\n" 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 ) except OSError: pass finally: client.close() thread = threading.Thread(target=serve_foreign, daemon=True) thread.start() process = subprocess.Popen( [str(CLI), "start"], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=dict(self.environment, FAKE_XRAY_LISTEN_PORT=str(wrong_port)), ) 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) marker = self.root / "foreign-listener-child-ran" results = [ self.run_cli("status"), self.run_cli("env"), self.run_cli("run", "--", "touch", str(marker)), self.run_cli("stop"), ] stdout, stderr = process.communicate(timeout=4) serving.clear() thread.join(timeout=1) self.assertTrue(all(result.returncode != 0 for result in results)) self.assertTrue(all("listener" in result.stderr for result in results)) self.assertFalse(marker.exists()) self.assertNotEqual(0, process.returncode) self.assertIn("listener", stderr) self.assert_no_secrets(stdout + stderr, profile_text) with socket.create_connection(("127.0.0.1", self.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()