ddadminandClaude Opus 5.5
09f57aa1e7
- state.json снова хранит адрес и порт: при смене конфига у работающего Xray status/env/run отказывают и советуют перезапустить start. - Ошибки разбора ссылки называют файл профиля. - Тесты: отказ start при чужой версии Xray, ключи REALITY не попадают в вывод, Xray узнаётся после замены файла; машинный конфиг проверяется через CLI; тест пути по умолчанию не зависит от доступности Hiddify. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
385 lines
15 KiB
Python
385 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Команда proxy-client с управляемым Xray: argv → код выхода, вывод, файлы, процессы."""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from urllib.parse import quote
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
CLI = REPO_ROOT / "proxy-client/.local/bin/proxy-client"
|
|
UUID = "11111111-2222-4333-8444-555555555555"
|
|
PATH_SECRET = "/tenant/PATH_SENTINEL_a1b2"
|
|
PROFILE = (
|
|
"vless://{}@cdn.example:8443?type=xhttp&security=tls&encryption=none"
|
|
"&sni=tls.example&fp=firefox&alpn=h2&host=host.example&path={}#Synthetic"
|
|
).format(UUID, quote(PATH_SECRET, safe=""))
|
|
PBK = "PUBLIC_KEY_SENTINEL_7f8a"
|
|
SID = "SHORT_ID_SENTINEL_9b3c"
|
|
REALITY_PROFILE = (
|
|
"vless://{}@edge.example:443?type=tcp&security=reality&encryption=none"
|
|
"&flow=xtls-rprx-vision&sni=front.example&fp=chrome&pbk={}&sid={}#Synthetic"
|
|
).format(UUID, PBK, SID)
|
|
READY = "остановка: Ctrl+C или proxy-client stop"
|
|
|
|
# Поддельный Xray — симлинк на интерпретатор Python: тогда /proc/PID/exe
|
|
# совпадает с разрешённым путём к xray, как у настоящего бинарника.
|
|
# «xray version» и «xray run ...» запускают одноимённые скрипты из рабочего
|
|
# каталога теста.
|
|
FAKE_VERSION = "print('Xray 26.3.27 (fake)')\n"
|
|
FAKE_RUN = r'''import json, os, socket, sys
|
|
|
|
config_path = sys.argv[-1]
|
|
if sys.argv[1] == "-test":
|
|
if os.environ.get("FAKE_TEST_FAIL"):
|
|
# Настоящий Xray может напечатать фрагменты конфига — CLI их глушит.
|
|
print(open(config_path).read())
|
|
sys.exit(1)
|
|
sys.exit(0)
|
|
|
|
inbound = json.load(open(config_path))["inbounds"][0]
|
|
server = socket.socket()
|
|
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
server.bind((inbound["listen"], inbound["port"]))
|
|
server.listen(16)
|
|
while True:
|
|
client, _ = server.accept()
|
|
request = client.recv(4096)
|
|
if b"ifconfig.me" in request and not os.environ.get("FAKE_IP_FAIL"):
|
|
client.sendall(
|
|
b"HTTP/1.1 200 OK\r\nContent-Length: 13\r\nConnection: close\r\n\r\n"
|
|
b"203.0.113.77\n"
|
|
)
|
|
client.close()
|
|
'''
|
|
|
|
|
|
def free_port():
|
|
with socket.socket() as listener:
|
|
listener.bind(("127.0.0.1", 0))
|
|
return listener.getsockname()[1]
|
|
|
|
|
|
def pid_alive(pid):
|
|
try:
|
|
os.kill(pid, 0)
|
|
except ProcessLookupError:
|
|
return False
|
|
return True
|
|
|
|
|
|
class ManagedXrayTest(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
self.root = Path(os.path.realpath(temporary.name))
|
|
self.addCleanup(self.kill_strays)
|
|
bin_dir = self.root / "bin"
|
|
bin_dir.mkdir()
|
|
self.xray = bin_dir / "xray"
|
|
self.xray.symlink_to(os.path.realpath(sys.executable))
|
|
(self.root / "version").write_text(FAKE_VERSION, encoding="utf-8")
|
|
(self.root / "run").write_text(FAKE_RUN, encoding="utf-8")
|
|
self.config_dir = self.root / "config/proxy-client"
|
|
self.state_dir = self.root / "state/proxy-client"
|
|
self.environment = dict(
|
|
os.environ,
|
|
HOME=str(self.root / "home"),
|
|
XDG_CONFIG_HOME=str(self.root / "config"),
|
|
XDG_STATE_HOME=str(self.root / "state"),
|
|
PATH=str(bin_dir) + os.pathsep + os.environ["PATH"],
|
|
PYTHONDONTWRITEBYTECODE="1",
|
|
)
|
|
self.port = free_port()
|
|
self.write_machine_config()
|
|
self.profile = self.write_profile(self.config_dir / "default.url")
|
|
|
|
def write_machine_config(self, address="127.0.0.1"):
|
|
self.config_dir.mkdir(parents=True, exist_ok=True)
|
|
(self.config_dir / "config").write_text(
|
|
"[proxy]\ntype = xray\nname = Managed test\naddress = {}\n"
|
|
"port = {}\nno_proxy_extra =\n".format(address, self.port),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def write_profile(self, path, mode=0o600, text=PROFILE):
|
|
path.write_text(text + "\n", encoding="utf-8")
|
|
path.chmod(mode)
|
|
return path
|
|
|
|
def run_cli(self, *arguments, **environment):
|
|
return subprocess.run(
|
|
[str(CLI)] + list(arguments),
|
|
cwd=str(self.root),
|
|
env=dict(self.environment, **environment),
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=15,
|
|
check=False,
|
|
)
|
|
|
|
def start(self, *arguments, **environment):
|
|
"""Запускает start и ждёт строку готовности; возвращает процесс."""
|
|
process = subprocess.Popen(
|
|
[str(CLI), "start"] + list(arguments),
|
|
cwd=str(self.root),
|
|
env=dict(self.environment, **environment),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
self.addCleanup(self.kill_leftovers, process)
|
|
watchdog = threading.Timer(10, process.kill)
|
|
watchdog.start()
|
|
self.start_output = ""
|
|
try:
|
|
for line in process.stdout:
|
|
self.start_output += line
|
|
if READY in line:
|
|
return process
|
|
finally:
|
|
watchdog.cancel()
|
|
self.fail("start не дошёл до готовности: " + process.communicate()[1])
|
|
|
|
def kill_strays(self):
|
|
# Все процессы теста (CLI и поддельный Xray) работают в self.root.
|
|
for entry in Path("/proc").iterdir():
|
|
try:
|
|
if os.readlink(str(entry / "cwd")) == str(self.root):
|
|
os.kill(int(entry.name), signal.SIGKILL)
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
def kill_leftovers(self, process):
|
|
self.kill_strays()
|
|
process.communicate()
|
|
|
|
def finish(self, process):
|
|
stdout, stderr = process.communicate(timeout=10)
|
|
return self.start_output + stdout, stderr
|
|
|
|
def xray_pid(self):
|
|
return json.loads((self.state_dir / "state.json").read_text(encoding="utf-8"))[
|
|
"xray_pid"
|
|
]
|
|
|
|
def assert_no_state_files(self):
|
|
self.assertFalse((self.state_dir / "state.json").exists())
|
|
self.assertFalse((self.state_dir / "client.json").exists())
|
|
|
|
def assert_no_secrets(self, text):
|
|
for secret in (UUID, PATH_SECRET, quote(PATH_SECRET, safe=""), PBK, SID):
|
|
self.assertNotIn(secret, text)
|
|
|
|
def test_full_lifecycle(self):
|
|
process = self.start()
|
|
xray_pid = self.xray_pid()
|
|
self.assertEqual(0o700, self.state_dir.stat().st_mode & 0o777)
|
|
for name in ("state.json", "client.json"):
|
|
self.assertEqual(0o600, (self.state_dir / name).stat().st_mode & 0o777)
|
|
|
|
status = self.run_cli("status")
|
|
env = self.run_cli("env")
|
|
child = self.run_cli(
|
|
"run", "--", sys.executable, "-c", "import os; print(os.environ['ALL_PROXY'])"
|
|
)
|
|
second = self.run_cli("start")
|
|
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:{}\n".format(self.port), env.stdout)
|
|
self.assertEqual("socks5://127.0.0.1:{}\n".format(self.port), child.stdout)
|
|
self.assertEqual(1, second.returncode)
|
|
self.assertIn("уже запущен", second.stderr)
|
|
|
|
stopped = self.run_cli("stop")
|
|
stdout, stderr = self.finish(process)
|
|
self.assertEqual(0, stopped.returncode, stopped.stderr)
|
|
self.assertEqual(0, process.returncode, stderr)
|
|
self.assertFalse(pid_alive(xray_pid))
|
|
self.assert_no_state_files()
|
|
self.assertIn("Внешний IP через прокси: 203.0.113.77", stdout)
|
|
self.assertIn("длина пути: {}".format(len(PATH_SECRET)), stdout)
|
|
self.assert_no_secrets(stdout + stderr + status.stdout + env.stdout)
|
|
|
|
def test_ctrl_c_stops_xray_and_removes_files(self):
|
|
process = self.start()
|
|
xray_pid = self.xray_pid()
|
|
process.send_signal(signal.SIGINT)
|
|
self.finish(process)
|
|
self.assertEqual(0, process.returncode)
|
|
self.assertFalse(pid_alive(xray_pid))
|
|
self.assert_no_state_files()
|
|
|
|
def test_stop_finishes_xray_left_after_killed_supervisor(self):
|
|
process = self.start()
|
|
xray_pid = self.xray_pid()
|
|
process.kill()
|
|
process.wait()
|
|
|
|
status = self.run_cli("status")
|
|
self.assertEqual(1, status.returncode)
|
|
self.assertIn("proxy-client stop", status.stderr)
|
|
restarted = self.run_cli("start")
|
|
self.assertEqual(1, restarted.returncode)
|
|
self.assertIn("proxy-client stop", restarted.stderr)
|
|
|
|
stopped = self.run_cli("stop")
|
|
self.assertEqual(0, stopped.returncode, stopped.stderr)
|
|
self.assertFalse(pid_alive(xray_pid))
|
|
self.assert_no_state_files()
|
|
|
|
def test_stale_files_after_processes_died_are_removed(self):
|
|
self.state_dir.mkdir(parents=True)
|
|
(self.state_dir / "state.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"supervisor_pid": 999999998,
|
|
"xray_pid": 999999999,
|
|
"xray_path": "/x",
|
|
"address": "127.0.0.1",
|
|
"port": self.port,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(self.state_dir / "client.json").write_text("{}", encoding="utf-8")
|
|
|
|
result = self.run_cli("status")
|
|
self.assertEqual(1, result.returncode)
|
|
self.assertIn("не запущен", result.stderr)
|
|
self.assert_no_state_files()
|
|
|
|
def test_broken_state_file_is_named_in_error(self):
|
|
self.state_dir.mkdir(parents=True)
|
|
state = self.state_dir / "state.json"
|
|
state.write_text("not json", encoding="utf-8")
|
|
result = self.run_cli("stop")
|
|
self.assertEqual(1, result.returncode)
|
|
self.assertIn("rm {}".format(state), result.stderr)
|
|
|
|
def test_busy_port_is_reported(self):
|
|
with socket.socket() as listener:
|
|
listener.bind(("127.0.0.1", self.port))
|
|
listener.listen()
|
|
result = self.run_cli("start")
|
|
self.assertEqual(1, result.returncode)
|
|
self.assertIn("порт {} занят".format(self.port), result.stderr)
|
|
self.assert_no_state_files()
|
|
|
|
def test_non_loopback_address_is_refused(self):
|
|
self.write_machine_config(address="192.0.2.10")
|
|
result = self.run_cli("start")
|
|
self.assertEqual(1, result.returncode)
|
|
self.assertIn("loopback", result.stderr)
|
|
|
|
def test_open_profile_names_target_file_and_gives_chmod_command(self):
|
|
target = self.write_profile(self.root / "real profile.url", mode=0o640)
|
|
link = self.root / "link.url"
|
|
link.symlink_to(target)
|
|
result = self.run_cli("start", "-f", str(link))
|
|
self.assertEqual(1, result.returncode)
|
|
self.assertIn(
|
|
"файл профиля {} доступен другим пользователям; "
|
|
"выполните: chmod 600 '{}'".format(target, target),
|
|
result.stderr,
|
|
)
|
|
self.assert_no_secrets(result.stdout + result.stderr)
|
|
|
|
def test_failed_config_test_leaves_no_files_and_no_secrets(self):
|
|
result = self.run_cli("start", FAKE_TEST_FAIL="1")
|
|
self.assertEqual(1, result.returncode)
|
|
self.assertIn("run -test", result.stderr)
|
|
self.assert_no_state_files()
|
|
self.assert_no_secrets(result.stdout + result.stderr)
|
|
|
|
def test_failed_ip_probe_is_only_a_warning(self):
|
|
process = self.start(FAKE_IP_FAIL="1")
|
|
self.assertIsNone(process.poll())
|
|
process.send_signal(signal.SIGINT)
|
|
_stdout, stderr = self.finish(process)
|
|
self.assertEqual(0, process.returncode)
|
|
self.assertIn("Предупреждение: внешний IP", stderr)
|
|
|
|
def test_reality_secrets_stay_out_of_start_output(self):
|
|
profile = self.write_profile(self.root / "reality.url", text=REALITY_PROFILE)
|
|
process = self.start("-f", str(profile))
|
|
process.send_signal(signal.SIGINT)
|
|
stdout, stderr = self.finish(process)
|
|
self.assertIn("Секреты REALITY: скрыты", stdout)
|
|
self.assert_no_secrets(stdout + stderr)
|
|
|
|
def test_wrong_xray_version_is_refused_before_launch(self):
|
|
(self.root / "version").write_text("print('Xray 26.3.26 (fake)')\n", encoding="utf-8")
|
|
result = self.run_cli("start")
|
|
self.assertEqual(1, result.returncode)
|
|
self.assertIn("не сообщил версию Xray 26.3.27", result.stderr)
|
|
self.assert_no_state_files()
|
|
|
|
def test_replaced_xray_file_is_still_recognised_and_stopped(self):
|
|
# Нужен настоящий файл, а не симлинк: только тогда замена файла видна
|
|
# в /proc/PID/exe как «(deleted)». Копия интерпретатора находит свою
|
|
# стандартную библиотеку через ../lib.
|
|
interpreter = Path(os.path.realpath(sys.executable))
|
|
self.xray.unlink()
|
|
shutil.copy2(str(interpreter), str(self.xray))
|
|
(self.root / "lib").symlink_to(interpreter.parent.parent / "lib")
|
|
process = self.start()
|
|
xray_pid = self.xray_pid()
|
|
|
|
replacement = self.xray.with_name("xray.new")
|
|
shutil.copy2(str(self.xray), str(replacement))
|
|
os.replace(str(replacement), str(self.xray))
|
|
self.assertTrue(os.readlink("/proc/{}/exe".format(xray_pid)).endswith(" (deleted)"))
|
|
|
|
status = self.run_cli("status")
|
|
self.assertEqual(0, status.returncode, status.stderr)
|
|
stopped = self.run_cli("stop")
|
|
self.assertEqual(0, stopped.returncode, stopped.stderr)
|
|
self.finish(process)
|
|
self.assertFalse(pid_alive(xray_pid))
|
|
self.assert_no_state_files()
|
|
|
|
def test_changed_machine_port_is_refused_while_xray_runs(self):
|
|
process = self.start()
|
|
managed_port = self.port
|
|
with socket.socket() as other:
|
|
other.bind(("127.0.0.1", 0))
|
|
other.listen()
|
|
self.port = other.getsockname()[1]
|
|
self.write_machine_config()
|
|
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(str(self.config_dir / "config"), result.stderr)
|
|
self.assertIn(":{}".format(managed_port), result.stderr)
|
|
self.assertIn("proxy-client stop", result.stderr)
|
|
self.assertFalse(marker.exists())
|
|
process.send_signal(signal.SIGINT)
|
|
self.finish(process)
|
|
|
|
def test_bad_link_error_names_profile_file(self):
|
|
target = self.root / "real profile.url"
|
|
target.write_text("vless://not-a-link\n", encoding="utf-8")
|
|
target.chmod(0o600)
|
|
link = self.root / "link.url"
|
|
link.symlink_to(target)
|
|
result = self.run_cli("start", "-f", str(link))
|
|
self.assertEqual(1, result.returncode)
|
|
self.assertIn("Ошибка: файл профиля {}: ".format(target), result.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|