fix(proxy-client): замечания слепого ревью

- 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>

proxy-client/.local/bin/proxy-client

@@ -215,7 +215,7 @@ def check_reachable(config):
 
 def check_proxy(config):
     if config.proxy_type == "xray":
-        require_supervised_xray()
+        require_supervised_xray(config)
     check_reachable(config)
 
 
@@ -262,7 +262,7 @@ def write_private_json(path, value):
     path.chmod(0o600)
 
 
-def read_profile(path):
+def load_profile(path):
     # Права проверяем у конечного файла: у симлинка они ничего не значат.
     resolved = path.expanduser().resolve(strict=True)
     if resolved.stat().st_mode & 0o077:
@@ -277,7 +277,10 @@ def read_profile(path):
         raise ProxyError(
             "файл профиля {} должен содержать ровно одну непустую VLESS-ссылку".format(resolved)
         )
-    return lines[0]
+    try:
+        return parse_vless(lines[0])
+    except ProxyError as error:
+        raise ProxyError("файл профиля {}: {}".format(resolved, error))
 
 
 def parse_query(raw_query):
@@ -581,7 +584,8 @@ def read_state():
     path = state_path()
     try:
         state = json.loads(path.read_text(encoding="utf-8"))
-        return {key: state[key] for key in ("supervisor_pid", "xray_pid", "xray_path")}
+        keys = ("supervisor_pid", "xray_pid", "xray_path", "address", "port")
+        return {key: state[key] for key in keys}
     except FileNotFoundError:
         raise ProxyError("управляемый Xray не запущен; выполните proxy-client start")
     except (ValueError, TypeError, KeyError):
@@ -634,12 +638,21 @@ def managed_xray_running():
         return False
 
 
-def require_supervised_xray():
+def require_supervised_xray(config):
-    if not supervisor_running(managed_state()):
+    state = managed_state()
+    if not supervisor_running(state):
         raise ProxyError(
             "Xray работает без proxy-client start (его процесс завершён); "
             "выполните proxy-client stop"
         )
+    if (config.address, config.port) != (state["address"], state["port"]):
+        raise ProxyError(
+            "Xray запущен на {}:{}, а {} указывает {}:{}; чтобы применить конфиг, "
+            "выполните proxy-client stop, затем proxy-client start".format(
+                state["address"], state["port"], config_home() / "config",
+                config.address, config.port,
+            )
+        )
 
 
 def wait_until(condition, timeout):
@@ -809,7 +822,7 @@ def run_managed_xray(config, profile_file):
         remove_state_files()
     check_port_available(ipaddress.ip_address(config.address), config.port)
     profile_path = Path(profile_file) if profile_file else config_home() / "default.url"
-    profile = parse_vless(read_profile(profile_path))
+    profile = load_profile(profile_path)
     executable = resolve_xray()
     check_xray_version(executable)
     write_private_json(
@@ -826,7 +839,13 @@ def run_managed_xray(config, profile_file):
         child = subprocess.Popen([executable, "run", "-c", str(generated_config_path())])
         write_private_json(
             state_path(),
-            {"supervisor_pid": os.getpid(), "xray_pid": child.pid, "xray_path": executable},
+            {
+                "supervisor_pid": os.getpid(),
+                "xray_pid": child.pid,
+                "xray_path": executable,
+                "address": config.address,
+                "port": config.port,
+            },
         )
         if not wait_until(
             lambda: port_open(config.address, config.port), max(1.0, preflight_timeout())

tests/test_proxy_client.py

@@ -152,8 +152,12 @@ class ExternalProxyCliTest(unittest.TestCase):
         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")
-        self.assertIn("Hiddify (hproxy.local:12334)", result.stderr)
+        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()
@@ -178,24 +182,37 @@ class ExternalProxyCliTest(unittest.TestCase):
         self.assertIn("положительным числом", result.stderr)
 
 
-class MachineConfigTest(unittest.TestCase):
+class MachineConfigCliTest(unittest.TestCase):
+    """Машинный конфиг проверяем через команду: код выхода и путь в ошибке."""
+
     VALID = (
         "[proxy]\ntype = external\nname = Test\naddress = 127.0.0.1\n"
         "port = 12334\nno_proxy_extra =\n"
     )
 
-    def parse(self, text):
+    def setUp(self):
-        return MODULE["parse_config"](text)
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name)
+        self.path = self.root / "proxy-client/config"
+        self.path.parent.mkdir()
 
-    def test_valid_config(self):
+    def start_with(self, text):
-        config = self.parse(self.VALID.replace("no_proxy_extra =", "no_proxy_extra = a, b"))
+        # start с type=external отказывает сразу после чтения конфига:
-        self.assertEqual(
+        # сеть и Xray не нужны.
-            ("external", "Test", "127.0.0.1", 12334, ("a", "b")),
+        self.path.write_text(text, encoding="utf-8")
-            (config.proxy_type, config.name, config.address, config.port, config.no_proxy_extra),
+        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_structure_and_values(self):
+    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",
@@ -210,33 +227,20 @@ class MachineConfigTest(unittest.TestCase):
                 "no_proxy_extra =", "no_proxy_extra = one.example,,two.example"
             ),
         }
-        for label, text in cases.items():
-            with self.subTest(label):
-                # Ошибки синтаксиса INI приходят от configparser; load_config
-                # добавляет к ним путь к файлу.
-                with self.assertRaises((ProxyError, MODULE["configparser"].Error)):
-                    self.parse(text)
-
-    def test_addresses(self):
-        for address in ("127.0.0.1", "2001:db8::1", "proxy-1.example", "trailing.example."):
-            with self.subTest(address):
-                self.assertEqual(
-                    address, self.parse(self.VALID.replace("127.0.0.1", address)).address
-                )
         for address in ("http://proxy", "[::1]", "has space", "proxy/path", "user@proxy",
                         "прокси.example", "fe80::1%eth0"):
-            with self.subTest(address):
+            cases["address " + address] = self.VALID.replace("127.0.0.1", address)
-                with self.assertRaisesRegex(ProxyError, "address"):
+        for label, text in cases.items():
-                    self.parse(self.VALID.replace("127.0.0.1", address))
+            with self.subTest(label):
+                result = self.start_with(text)
+                self.assertEqual(1, result.returncode)
+                self.assertIn("Ошибка: машинный конфиг {}:".format(self.path), result.stderr)
 
-    def test_load_error_names_config_file(self):
+    def test_valid_addresses_are_accepted(self):
-        with tempfile.TemporaryDirectory() as root:
+        for address in ("127.0.0.1", "2001:db8::1", "proxy-1.example", "trailing.example."):
-            path = Path(root) / "proxy-client/config"
+            with self.subTest(address):
-            path.parent.mkdir()
+                result = self.start_with(self.VALID.replace("127.0.0.1", address))
-            path.write_text("not an ini\n", encoding="utf-8")
+                self.assertIn("только для type=xray", result.stderr)
-            with mock.patch.dict(os.environ, {"XDG_CONFIG_HOME": root}):
-                with self.assertRaisesRegex(ProxyError, str(path)):
-                    MODULE["load_config"]()
 
 
 UUID = "11111111-2222-4333-8444-555555555555"
@@ -538,6 +542,8 @@ class XrayInstallTest(unittest.TestCase):
                     "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",

tests/test_proxy_client_managed.py

@@ -2,6 +2,7 @@
 """Команда proxy-client с управляемым Xray: argv → код выхода, вывод, файлы, процессы."""
 import json
 import os
+import shutil
 import signal
 import socket
 import subprocess
@@ -21,6 +22,12 @@ 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
@@ -73,7 +80,8 @@ class ManagedXrayTest(unittest.TestCase):
     def setUp(self):
         temporary = tempfile.TemporaryDirectory()
         self.addCleanup(temporary.cleanup)
-        self.root = Path(temporary.name)
+        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"
@@ -102,8 +110,8 @@ class ManagedXrayTest(unittest.TestCase):
             encoding="utf-8",
         )
 
-    def write_profile(self, path, mode=0o600):
+    def write_profile(self, path, mode=0o600, text=PROFILE):
-        path.write_text(PROFILE + "\n", encoding="utf-8")
+        path.write_text(text + "\n", encoding="utf-8")
         path.chmod(mode)
         return path
 
@@ -118,10 +126,10 @@ class ManagedXrayTest(unittest.TestCase):
             check=False,
         )
 
-    def start(self, **environment):
+    def start(self, *arguments, **environment):
         """Запускает start и ждёт строку готовности; возвращает процесс."""
         process = subprocess.Popen(
-            [str(CLI), "start"],
+            [str(CLI), "start"] + list(arguments),
             cwd=str(self.root),
             env=dict(self.environment, **environment),
             text=True,
@@ -141,14 +149,17 @@ class ManagedXrayTest(unittest.TestCase):
             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):
-        state = self.state_dir / "state.json"
+        self.kill_strays()
-        if state.exists():
-            pid = json.loads(state.read_text(encoding="utf-8"))["xray_pid"]
-            if pid_alive(pid):
-                os.kill(pid, signal.SIGKILL)
-        if process.poll() is None:
-            process.kill()
         process.communicate()
 
     def finish(self, process):
@@ -165,7 +176,7 @@ class ManagedXrayTest(unittest.TestCase):
         self.assertFalse((self.state_dir / "client.json").exists())
 
     def assert_no_secrets(self, text):
-        for secret in (UUID, PATH_SECRET, quote(PATH_SECRET, safe="")):
+        for secret in (UUID, PATH_SECRET, quote(PATH_SECRET, safe=""), PBK, SID):
             self.assertNotIn(secret, text)
 
     def test_full_lifecycle(self):
@@ -230,7 +241,13 @@ class ManagedXrayTest(unittest.TestCase):
         self.state_dir.mkdir(parents=True)
         (self.state_dir / "state.json").write_text(
             json.dumps(
-                {"supervisor_pid": 999999998, "xray_pid": 999999999, "xray_path": "/x"}
+                {
+                    "supervisor_pid": 999999998,
+                    "xray_pid": 999999999,
+                    "xray_path": "/x",
+                    "address": "127.0.0.1",
+                    "port": self.port,
+                }
             ),
             encoding="utf-8",
         )
@@ -292,6 +309,76 @@ class ManagedXrayTest(unittest.TestCase):
         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()