feat(xray-proxy): добавлена проверяемая установка Xray командой install

- Зачем:
  - пользователь без root должен получать проверенный бинарник Xray 26.3.27
    одной явной командой, без риска для живого managed-экземпляра.
- Что:
  - команда install [--replace]: пин 26.3.27, маппинг x86_64/aarch64,
    скачивание urllib + проверка SHA2-256 из официального четырёхзаписного
    .dgst, staging внутри ~/.local/bin и один атомарный os.replace;
  - границы размеров (64 MiB архив/бинарник, 4 KiB .dgst, счёт limit+1,
    сверка Content-Length) и guard managed-режима через start-lock;
  - типизированная StartLockBusyError, offline-тесты реального загрузчика
    через подмену urlopen, снапшоты дерева ФС в отказных путях;
  - README/спека обновлены, включая справку о последней версии с
    allowInsecure (25.12.8, удалена в 26.2.6); тикет 04 закрыт, леджер
    дополнен итогами эксперимента.
- Проверка:
  - PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s tests — 79 OK.
  - bash tests/test_proxy_launchers.sh — 8 PASS.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_xray_proxy.py

@@ -1,11 +1,16 @@
 #!/usr/bin/env python3
+import fcntl
+import http.client
 import json
+import hashlib
 import os
 import runpy
 import subprocess
 import sys
 import tempfile
 import unittest
+import urllib.error
+import zipfile
 from unittest import mock
 from pathlib import Path
 
@@ -14,6 +19,31 @@ REPO_ROOT = Path(__file__).resolve().parents[1]
 CLI = REPO_ROOT / "xray-proxy/.local/bin/xray-proxy"
 
 
+class FakeUrlResponse:
+    def __init__(self, data=b"", content_length=None, read_error=None):
+        self.data = data
+        self.position = 0
+        self.read_error = read_error
+        self.headers = {}
+        if content_length is not None:
+            self.headers["Content-Length"] = str(content_length)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, _type, _value, _traceback):
+        return False
+
+    def read(self, size=-1):
+        if self.read_error is not None:
+            raise self.read_error
+        if size < 0:
+            size = len(self.data) - self.position
+        chunk = self.data[self.position : self.position + size]
+        self.position += len(chunk)
+        return chunk
+
+
 def checking_socket_setup(outcome="success"):
     if outcome == "success":
         outcome_line = "    return Connection()\n"
@@ -517,5 +547,689 @@ class XrayProxyCliTest(unittest.TestCase):
                 self.assertIn("address", result.stderr)
 
 
+class XrayInstallTest(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary.name)
+        self.home = self.root / "home"
+        self.state_home = self.root / "state"
+        self.home.mkdir()
+        self.environment = {
+            "HOME": str(self.home),
+            "XDG_STATE_HOME": str(self.state_home),
+            "PATH": str(self.home / ".local/bin"),
+        }
+        self.module = runpy.run_path(str(CLI))
+        self.destination = self.home / ".local/bin/xray"
+        self.archive = self.root / "fixture.zip"
+        self.write_archive("installed")
+
+    def tearDown(self):
+        self.temporary.cleanup()
+
+    @staticmethod
+    def fake_xray(label="fixture", version="26.3.27"):
+        return (
+            "#!/bin/sh\n"
+            "if [ \"$1\" = version ]; then\n"
+            "  printf '%s\\n' 'Xray {} ({})'\n"
+            "  exit 0\n"
+            "fi\n"
+            "exit 64\n".format(version, label)
+        ).encode("ascii")
+
+    def write_archive(self, label="fixture", version="26.3.27", member="xray"):
+        with zipfile.ZipFile(str(self.archive), "w") as archive:
+            archive.writestr(member, self.fake_xray(label, version))
+
+    @staticmethod
+    def official_digest(archive_hash):
+        return (
+            "MD5= {}\n"
+            "SHA1= {}\n"
+            "SHA2-256= {}\n"
+            "SHA2-512= {}\n".format(
+                "1" * 32,
+                "2" * 40,
+                archive_hash,
+                "3" * 128,
+            )
+        )
+
+    def downloader(self, bad_hash=False, digest_text=None, fail_after_archive=False):
+        calls = []
+
+        def download(url, destination, _max_bytes):
+            calls.append(url)
+            if url.endswith(".dgst"):
+                if fail_after_archive:
+                    raise self.module["ProxyError"]("инъецированная ошибка загрузки")
+                archive_hash = hashlib.sha256(self.archive.read_bytes()).hexdigest()
+                if bad_hash:
+                    archive_hash = "0" * 64
+                text = digest_text
+                if text is None:
+                    text = self.official_digest(archive_hash)
+                destination.write_text(text, encoding="ascii")
+            else:
+                destination.write_bytes(self.archive.read_bytes())
+
+        return download, calls
+
+    def install(self, replace=False, downloader=None, machine="x86_64", path=None):
+        if downloader is None:
+            downloader, _calls = self.downloader()
+        environment = dict(self.environment)
+        if path is not None:
+            environment["PATH"] = path
+        globals_ = self.module["command_install"].__globals__
+        with mock.patch.dict(os.environ, environment, clear=False):
+            with mock.patch.object(globals_["platform"], "machine", return_value=machine):
+                return self.module["command_install"](replace, downloader)
+
+    def write_destination(self, content=None, executable=True):
+        self.destination.parent.mkdir(parents=True, exist_ok=True)
+        self.destination.write_bytes(content or self.fake_xray("existing"))
+        self.destination.chmod(0o755 if executable else 0o644)
+
+    def tree_snapshot(self):
+        snapshot = {}
+        for path in self.root.rglob("*"):
+            relative = path.relative_to(self.root).as_posix()
+            info = path.lstat()
+            mode = info.st_mode & 0o777
+            if path.is_symlink():
+                snapshot[relative] = ("symlink", mode, os.readlink(str(path)))
+            elif path.is_dir():
+                snapshot[relative] = ("directory", mode)
+            elif path.is_file():
+                digest = hashlib.sha256(path.read_bytes()).hexdigest()
+                snapshot[relative] = ("file", mode, digest)
+            else:
+                snapshot[relative] = ("other", mode)
+        return snapshot
+
+    def tree_snapshot_without_install_lock(self):
+        snapshot = self.tree_snapshot()
+        for path in ("state/xray-proxy/lock", "state/xray-proxy", "state"):
+            snapshot.pop(path, None)
+        return snapshot
+
+    def assert_no_staging(self):
+        if self.destination.parent.exists():
+            self.assertEqual(
+                [], list(self.destination.parent.glob(".xray-install-*"))
+            )
+
+    def assert_install_lock_released(self):
+        lock_path = self.state_home / "xray-proxy/lock"
+        with lock_path.open("r+", encoding="ascii") as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+    def test_architecture_mapping_is_pure_and_rejects_unknown(self):
+        asset_for = self.module["xray_asset_for_architecture"]
+        for machine in ("x86_64", "amd64"):
+            self.assertEqual("Xray-linux-64.zip", asset_for(machine))
+        for machine in ("aarch64", "arm64"):
+            self.assertEqual("Xray-linux-arm64-v8a.zip", asset_for(machine))
+        with self.assertRaisesRegex(self.module["ProxyError"], "не поддерживается"):
+            asset_for("riscv64")
+
+    def test_valid_digest_installs_verified_executable_and_only_expected_files(self):
+        downloader, calls = self.downloader()
+        self.assertEqual(0, self.install(downloader=downloader))
+
+        self.assertEqual(0o755, self.destination.stat().st_mode & 0o777)
+        self.assertIn(b"installed", self.destination.read_bytes())
+        self.assertEqual(2, len(calls))
+        self.assertTrue(calls[0].endswith("/Xray-linux-64.zip"))
+        self.assertEqual(calls[0] + ".dgst", calls[1])
+        self.assertEqual(
+            {
+                "fixture.zip",
+                "home",
+                "home/.local",
+                "home/.local/bin",
+                "home/.local/bin/xray",
+                "state",
+                "state/xray-proxy",
+                "state/xray-proxy/lock",
+            },
+            set(self.tree_snapshot()),
+        )
+        self.assert_no_staging()
+
+    def test_platform_machine_selects_arm64_asset(self):
+        downloader, calls = self.downloader()
+        self.assertEqual(
+            0, self.install(downloader=downloader, machine="aarch64")
+        )
+        self.assertTrue(calls[0].endswith("/Xray-linux-arm64-v8a.zip"))
+        self.assertEqual(calls[0] + ".dgst", calls[1])
+
+    def test_candidate_is_private_while_zip_bytes_are_written(self):
+        os_module = self.module["extract_xray_candidate"].__globals__["os"]
+        real_fdopen = os_module.fdopen
+        observed_modes = []
+
+        def checked_fdopen(descriptor, *arguments, **keywords):
+            descriptor_path = os.readlink("/proc/self/fd/{}".format(descriptor))
+            if Path(descriptor_path).name == "xray":
+                observed_modes.append(os.fstat(descriptor).st_mode & 0o777)
+            return real_fdopen(descriptor, *arguments, **keywords)
+
+        with mock.patch.object(
+            os_module, "fdopen", side_effect=checked_fdopen
+        ):
+            self.assertEqual(0, self.install())
+
+        self.assertEqual([0o600], observed_modes)
+        self.assertEqual(0o755, self.destination.stat().st_mode & 0o777)
+
+    def test_bad_hash_and_digest_format_preserve_destination_before_extraction(self):
+        valid_hash = "a" * 64
+        cases = (
+            self.downloader(bad_hash=True)[0],
+            self.downloader(digest_text="SHA256= " + valid_hash + "\n")[0],
+            self.downloader(
+                digest_text=self.official_digest(valid_hash).replace(
+                    "SHA2-256= " + valid_hash + "\n", ""
+                )
+            )[0],
+            self.downloader(
+                digest_text=self.official_digest(valid_hash)
+                + "SHA2-256= "
+                + valid_hash
+                + "\n"
+            )[0],
+            self.downloader(
+                digest_text=self.official_digest(valid_hash) + "garbage\n"
+            )[0],
+            self.downloader(
+                digest_text=self.official_digest(valid_hash).replace(
+                    valid_hash, "a" * 63
+                )
+            )[0],
+            self.downloader(
+                digest_text=self.official_digest(valid_hash).replace(
+                    valid_hash, "a" * 65
+                )
+            )[0],
+        )
+        for downloader in cases:
+            with self.subTest(downloader=downloader):
+                sentinel = b"destination sentinel"
+                self.write_destination(sentinel)
+                before = self.tree_snapshot_without_install_lock()
+                with self.assertRaises(self.module["ProxyError"]):
+                    self.install(replace=True, downloader=downloader)
+                self.assertEqual(sentinel, self.destination.read_bytes())
+                self.assert_no_staging()
+                self.assertEqual(before, self.tree_snapshot_without_install_lock())
+
+    def test_bad_or_truncated_archive_is_rejected_before_zip_open(self):
+        full_archive = self.archive.read_bytes()
+
+        def truncated_downloader(url, destination, _max_bytes):
+            if url.endswith(".dgst"):
+                digest = hashlib.sha256(full_archive).hexdigest()
+                destination.write_text(self.official_digest(digest), encoding="ascii")
+            else:
+                destination.write_bytes(full_archive[: len(full_archive) // 2])
+
+        zip_file = self.module["extract_xray_candidate"].__globals__["zipfile"]
+        with mock.patch.object(
+            zip_file, "ZipFile", side_effect=AssertionError("ZIP открыт до SHA-256")
+        ) as opened:
+            with self.assertRaisesRegex(self.module["ProxyError"], "SHA-256"):
+                self.install(downloader=truncated_downloader)
+        opened.assert_not_called()
+        self.assertFalse(self.destination.exists())
+        self.assert_no_staging()
+
+    def test_real_downloader_installs_complete_declared_responses(self):
+        archive = self.archive.read_bytes()
+        digest = self.official_digest(hashlib.sha256(archive).hexdigest()).encode(
+            "ascii"
+        )
+        responses = [
+            FakeUrlResponse(archive, content_length=len(archive)),
+            FakeUrlResponse(digest, content_length=len(digest)),
+        ]
+        urllib_request = self.module["download_file"].__globals__["urllib"].request
+
+        with mock.patch.object(urllib_request, "urlopen", side_effect=responses):
+            self.assertEqual(
+                0, self.install(downloader=self.module["download_file"])
+            )
+
+    def test_real_downloader_refuses_length_errors_and_releases_install(self):
+        archive = self.archive.read_bytes()
+        cases = (
+            (
+                "early EOF",
+                FakeUrlResponse(archive, content_length=len(archive) + 1),
+            ),
+            (
+                "extra bytes",
+                FakeUrlResponse(archive, content_length=len(archive) - 1),
+            ),
+            (
+                "declared oversized",
+                FakeUrlResponse(
+                    b"",
+                    content_length=self.module["MAX_ARCHIVE_BYTES"] + 1,
+                ),
+            ),
+            ("url error", urllib.error.URLError("offline")),
+            (
+                "http error",
+                FakeUrlResponse(read_error=http.client.HTTPException("broken HTTP")),
+            ),
+            (
+                "incomplete read",
+                FakeUrlResponse(
+                    read_error=http.client.IncompleteRead(b"partial", 10)
+                ),
+            ),
+        )
+        urllib_request = self.module["download_file"].__globals__["urllib"].request
+
+        for label, outcome in cases:
+            with self.subTest(label=label):
+                sentinel = b"destination sentinel"
+                self.write_destination(sentinel)
+                before = self.tree_snapshot_without_install_lock()
+                with mock.patch.object(urllib_request, "urlopen", side_effect=[outcome]):
+                    with self.assertRaises(self.module["ProxyError"]):
+                        self.install(
+                            replace=True, downloader=self.module["download_file"]
+                        )
+                self.assertEqual(sentinel, self.destination.read_bytes())
+                self.assert_no_staging()
+                self.assert_install_lock_released()
+                self.assertEqual(before, self.tree_snapshot_without_install_lock())
+
+    def test_real_downloader_counts_limit_plus_one_without_content_length(self):
+        urllib_request = self.module["download_file"].__globals__["urllib"].request
+        globals_ = self.module["command_install"].__globals__
+        sentinel = b"destination sentinel"
+        self.write_destination(sentinel)
+
+        with mock.patch.dict(globals_, {"MAX_ARCHIVE_BYTES": 8}):
+            response = FakeUrlResponse(b"123456789")
+            with mock.patch.object(urllib_request, "urlopen", return_value=response):
+                with self.assertRaisesRegex(self.module["ProxyError"], "размер"):
+                    self.install(
+                        replace=True, downloader=self.module["download_file"]
+                    )
+
+        self.assertEqual(sentinel, self.destination.read_bytes())
+        self.assert_no_staging()
+        self.assert_install_lock_released()
+
+    def test_uppercase_digest_is_rejected_before_zip_open(self):
+        digest = hashlib.sha256(self.archive.read_bytes()).hexdigest().upper()
+        downloader, _calls = self.downloader(
+            digest_text=self.official_digest(digest)
+        )
+        zip_file = self.module["extract_xray_candidate"].__globals__["zipfile"]
+        with mock.patch.object(
+            zip_file, "ZipFile", side_effect=AssertionError("ZIP открыт до digest")
+        ) as opened:
+            with self.assertRaisesRegex(self.module["ProxyError"], "SHA-256"):
+                self.install(downloader=downloader)
+        opened.assert_not_called()
+        self.assertFalse(self.destination.exists())
+        self.assert_no_staging()
+
+    def test_existing_verified_install_is_noop_without_downloader(self):
+        self.write_destination()
+        downloader = mock.Mock(side_effect=AssertionError("network used"))
+        self.assertEqual(0, self.install(downloader=downloader))
+        downloader.assert_not_called()
+
+    def test_existing_version_gate_keeps_timeout_and_avoids_downloader(self):
+        self.write_destination()
+        downloader = mock.Mock(side_effect=AssertionError("network used"))
+        subprocess_module = self.module["xray_version"].__globals__["subprocess"]
+        timeout = subprocess.TimeoutExpired([str(self.destination), "version"], 5)
+
+        with mock.patch.object(subprocess_module, "run", side_effect=timeout) as run:
+            with self.assertRaisesRegex(self.module["ProxyError"], "--replace"):
+                self.install(downloader=downloader)
+
+        downloader.assert_not_called()
+        self.assertEqual(5, run.call_args.kwargs["timeout"])
+        self.assertEqual(self.fake_xray("existing"), self.destination.read_bytes())
+
+    def test_replace_reinstalls_verified_existing_file(self):
+        self.write_destination(self.fake_xray("old"))
+        downloader, calls = self.downloader()
+        self.assertEqual(0, self.install(replace=True, downloader=downloader))
+        self.assertIn(b"installed", self.destination.read_bytes())
+        self.assertEqual(2, len(calls))
+
+    def test_destination_table_for_symlink_directory_and_bad_files(self):
+        self.destination.parent.mkdir(parents=True)
+        target = self.root / "target"
+        target.write_bytes(b"target sentinel")
+        self.destination.symlink_to(target)
+        with self.assertRaisesRegex(self.module["ProxyError"], "--replace"):
+            self.install()
+        self.assertTrue(self.destination.is_symlink())
+        self.assertEqual(0, self.install(replace=True))
+        self.assertFalse(self.destination.is_symlink())
+        self.assertEqual(b"target sentinel", target.read_bytes())
+
+        self.destination.unlink()
+        self.destination.mkdir()
+        for replace in (False, True):
+            with self.subTest(directory=True, replace=replace):
+                with self.assertRaisesRegex(self.module["ProxyError"], "каталог"):
+                    self.install(replace=replace)
+        self.destination.rmdir()
+
+        for content, executable in ((b"broken", True), (self.fake_xray(), False)):
+            with self.subTest(content=content, executable=executable):
+                self.write_destination(content, executable)
+                with self.assertRaisesRegex(self.module["ProxyError"], "--replace"):
+                    self.install()
+                self.assertEqual(0, self.install(replace=True))
+                self.assertIn(b"installed", self.destination.read_bytes())
+                self.destination.unlink()
+
+    def test_parent_symlink_cannot_redirect_install_outside_home(self):
+        outside = self.root / "outside"
+        outside.mkdir()
+        (self.home / ".local").symlink_to(outside, target_is_directory=True)
+        downloader = mock.Mock()
+
+        with self.assertRaisesRegex(self.module["ProxyError"], "домашнего каталога"):
+            self.install(downloader=downloader)
+
+        downloader.assert_not_called()
+        self.assertEqual([], list(outside.iterdir()))
+
+    def test_managed_state_guard_only_allows_absent_or_stale(self):
+        guard = self.module["require_install_safe_managed_state"]
+        globals_ = guard.__globals__
+        present = mock.Mock()
+        present.lstat.return_value = mock.Mock()
+        with mock.patch.dict(
+            globals_,
+            {
+                "state_path": mock.Mock(return_value=present),
+                "read_managed_state": mock.Mock(return_value={"state": "fixture"}),
+            },
+        ):
+            for outcome in ("healthy", "orphan", "unverifiable"):
+                with self.subTest(outcome=outcome):
+                    globals_["managed_state_outcome"] = mock.Mock(return_value=outcome)
+                    with self.assertRaisesRegex(self.module["ProxyError"], "stop"):
+                        guard()
+            globals_["managed_state_outcome"] = mock.Mock(return_value="stale")
+            guard()
+
+        absent = mock.Mock()
+        absent.lstat.side_effect = FileNotFoundError()
+        with mock.patch.dict(
+            globals_, {"state_path": mock.Mock(return_value=absent)}
+        ):
+            guard()
+
+    def test_command_classifies_managed_state_before_even_noop_destination(self):
+        self.write_destination()
+        downloader = mock.Mock(side_effect=AssertionError("network used"))
+        guard = mock.Mock(
+            side_effect=self.module["ProxyError"](
+                "managed Xray работает; сначала выполните xray-proxy stop"
+            )
+        )
+        globals_ = self.module["command_install"].__globals__
+
+        with mock.patch.dict(
+            globals_, {"require_install_safe_managed_state": guard}
+        ):
+            with self.assertRaisesRegex(self.module["ProxyError"], "stop"):
+                self.install(downloader=downloader)
+
+        guard.assert_called_once_with()
+        downloader.assert_not_called()
+        self.assertEqual(self.fake_xray("existing"), self.destination.read_bytes())
+
+    def test_download_extraction_and_version_errors_are_atomic_and_clean_staging(self):
+        failures = []
+        failures.append(self.downloader(fail_after_archive=True)[0])
+        missing_member = self.root / "missing-member.zip"
+        with zipfile.ZipFile(str(missing_member), "w") as archive:
+            archive.writestr("not-xray", b"data")
+
+        def missing_member_downloader(url, destination, _max_bytes):
+            source = missing_member if not url.endswith(".dgst") else None
+            if source is not None:
+                destination.write_bytes(source.read_bytes())
+            else:
+                digest = hashlib.sha256(missing_member.read_bytes()).hexdigest()
+                destination.write_text(self.official_digest(digest), encoding="ascii")
+
+        failures.append(missing_member_downloader)
+        self.write_archive("wrong", version="26.3.26")
+        failures.append(self.downloader()[0])
+
+        for downloader in failures:
+            with self.subTest(downloader=downloader):
+                sentinel = b"destination sentinel"
+                self.write_destination(sentinel)
+                before = self.tree_snapshot_without_install_lock()
+                with self.assertRaises(self.module["ProxyError"]):
+                    self.install(replace=True, downloader=downloader)
+                self.assertEqual(sentinel, self.destination.read_bytes())
+                self.assert_no_staging()
+                self.assertEqual(before, self.tree_snapshot_without_install_lock())
+
+    def test_corrupt_zip_with_matching_digest_is_atomic_and_cleaned(self):
+        corrupt = b"not a ZIP archive"
+
+        def corrupt_downloader(url, destination, _max_bytes):
+            if url.endswith(".dgst"):
+                digest = hashlib.sha256(corrupt).hexdigest()
+                destination.write_text(self.official_digest(digest), encoding="ascii")
+            else:
+                destination.write_bytes(corrupt)
+
+        sentinel = b"destination sentinel"
+        self.write_destination(sentinel)
+        before = self.tree_snapshot_without_install_lock()
+        with self.assertRaisesRegex(self.module["ProxyError"], "извлечь"):
+            self.install(replace=True, downloader=corrupt_downloader)
+        self.assertEqual(sentinel, self.destination.read_bytes())
+        self.assert_no_staging()
+        self.assertEqual(before, self.tree_snapshot_without_install_lock())
+
+    def test_injected_oversized_archive_and_digest_are_bounded(self):
+        globals_ = self.module["command_install"].__globals__
+        cases = (
+            ("MAX_ARCHIVE_BYTES", len(self.archive.read_bytes()) - 1, self.downloader()[0]),
+            (
+                "MAX_DIGEST_BYTES",
+                8,
+                self.downloader(digest_text=self.official_digest("a" * 64))[0],
+            ),
+        )
+        for constant, limit, downloader in cases:
+            with self.subTest(constant=constant):
+                sentinel = b"destination sentinel"
+                self.write_destination(sentinel)
+                with mock.patch.dict(globals_, {constant: limit}):
+                    with self.assertRaisesRegex(self.module["ProxyError"], "размер"):
+                        self.install(replace=True, downloader=downloader)
+                self.assertEqual(sentinel, self.destination.read_bytes())
+                self.assert_no_staging()
+
+    def test_zip_member_declared_over_limit_is_rejected_before_extraction(self):
+        globals_ = self.module["command_install"].__globals__
+        binary_size = len(self.fake_xray())
+        sentinel = b"destination sentinel"
+        self.write_destination(sentinel)
+
+        with mock.patch.dict(globals_, {"MAX_XRAY_BYTES": binary_size - 1}):
+            with self.assertRaisesRegex(self.module["ProxyError"], "размер"):
+                self.install(replace=True)
+
+        self.assertEqual(sentinel, self.destination.read_bytes())
+        self.assert_no_staging()
+
+    def test_streamed_zip_output_cannot_exceed_limit_even_if_metadata_lies(self):
+        class FakeInfo:
+            filename = "xray"
+            file_size = 8
+
+            @staticmethod
+            def is_dir():
+                return False
+
+        class FakeArchive:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, _type, _value, _traceback):
+                return False
+
+            @staticmethod
+            def infolist():
+                return [FakeInfo()]
+
+            @staticmethod
+            def open(_member):
+                return FakeUrlResponse(b"123456789")
+
+        globals_ = self.module["command_install"].__globals__
+        zip_file = globals_["zipfile"]
+        sentinel = b"destination sentinel"
+        self.write_destination(sentinel)
+        with mock.patch.dict(globals_, {"MAX_XRAY_BYTES": 8}):
+            with mock.patch.object(zip_file, "ZipFile", return_value=FakeArchive()):
+                with self.assertRaisesRegex(self.module["ProxyError"], "размер"):
+                    self.install(replace=True)
+        self.assertEqual(sentinel, self.destination.read_bytes())
+        self.assert_no_staging()
+
+    def test_interruption_preserves_destination_cleans_staging_and_releases_lock(self):
+        sentinel = b"destination sentinel"
+        self.write_destination(sentinel)
+        before = self.tree_snapshot_without_install_lock()
+
+        def interrupted_downloader(_url, destination, _max_bytes):
+            destination.write_bytes(b"partial")
+            raise KeyboardInterrupt()
+
+        with self.assertRaises(KeyboardInterrupt):
+            self.install(replace=True, downloader=interrupted_downloader)
+
+        self.assertEqual(sentinel, self.destination.read_bytes())
+        self.assert_no_staging()
+        lock_path = self.state_home / "xray-proxy/lock"
+        with lock_path.open("r+", encoding="ascii") as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        self.assertEqual(before, self.tree_snapshot_without_install_lock())
+
+    def test_start_lock_is_held_during_download_and_through_replace(self):
+        downloader, _calls = self.downloader()
+        lock_path = self.state_home / "xray-proxy/lock"
+
+        def assert_lock_held():
+            with lock_path.open("r+", encoding="ascii") as competing:
+                with self.assertRaises(BlockingIOError):
+                    fcntl.flock(competing, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+        def checked_downloader(url, destination, max_bytes):
+            assert_lock_held()
+            downloader(url, destination, max_bytes)
+
+        os_module = self.module["command_install"].__globals__["os"]
+        real_replace = os_module.replace
+
+        def checked_replace(source, destination):
+            real_replace(source, destination)
+            assert_lock_held()
+
+        with mock.patch.object(os_module, "replace", side_effect=checked_replace):
+            self.assertEqual(0, self.install(downloader=checked_downloader))
+
+        with lock_path.open("r+", encoding="ascii") as released:
+            fcntl.flock(released, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+    def test_unsupported_architecture_refuses_before_downloading(self):
+        downloader = mock.Mock()
+        with self.assertRaisesRegex(self.module["ProxyError"], "не поддерживается"):
+            self.install(downloader=downloader, machine="riscv64")
+        downloader.assert_not_called()
+        self.assertFalse(self.destination.exists())
+
+    def test_path_missing_or_competing_returns_error_without_rollback(self):
+        cases = (("empty", False), ("competitor", True))
+        for directory_name, competing in cases:
+            with self.subTest(directory_name=directory_name):
+                path_dir = self.root / directory_name
+                path_dir.mkdir()
+                if competing:
+                    competitor = path_dir / "xray"
+                    competitor.write_bytes(self.fake_xray("competitor"))
+                    competitor.chmod(0o755)
+                    path = str(path_dir) + os.pathsep + str(self.destination.parent)
+                else:
+                    path = str(path_dir)
+                with self.assertRaisesRegex(self.module["ProxyError"], "PATH"):
+                    self.install(path=path)
+                self.assertTrue(self.destination.is_file())
+                self.destination.unlink()
+
+    def test_busy_install_hint_uses_exception_type_not_message_text(self):
+        globals_ = self.module["acquire_install_lock"].__globals__
+        busy_error = self.module["StartLockBusyError"]("completely changed text")
+        with mock.patch.dict(
+            globals_, {"acquire_start_lock": mock.Mock(side_effect=busy_error)}
+        ):
+            with self.assertRaisesRegex(self.module["ProxyError"], "stop"):
+                self.module["acquire_install_lock"]()
+
+    def test_path_comparison_handles_relative_entry_and_internal_bin_symlink(self):
+        local = self.home / ".local"
+        actual_bin = self.home / "private-bin"
+        local.mkdir()
+        actual_bin.mkdir()
+        (local / "bin").symlink_to(actual_bin, target_is_directory=True)
+
+        self.assertEqual(0, self.install())
+        self.assertTrue((actual_bin / "xray").is_file())
+
+        previous_cwd = Path.cwd()
+        try:
+            os.chdir(str(actual_bin))
+            self.assertEqual(0, self.install(path="."))
+        finally:
+            os.chdir(str(previous_cwd))
+
+    def test_install_dispatches_before_machine_config_loading(self):
+        self.write_destination()
+        config = self.root / "config/proxy"
+        config.mkdir(parents=True)
+        (config / "config").write_text("not an ini\n", encoding="ascii")
+        environment = os.environ.copy()
+        environment.update(self.environment)
+        environment["PATH"] = self.environment["PATH"] + os.pathsep + os.environ["PATH"]
+        environment["XDG_CONFIG_HOME"] = str(self.root / "config")
+        result = subprocess.run(
+            [str(CLI), "install"],
+            text=True,
+            capture_output=True,
+            env=environment,
+            check=False,
+        )
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertIn("уже установлен", result.stdout)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_xray_proxy_managed.py

@@ -473,6 +473,24 @@ class ManagedXrayCliTest(unittest.TestCase):
             profile_text,
         )
 
+    def test_live_managed_instance_rejects_install_replace_without_changes(self):
+        profile_text = self.tcp_profile()
+        self.write_profile(profile_text)
+        process = self.start()
+        state_before = self.managed_state_path().read_bytes()
+        xray_before = self.xray.read_bytes()
+
+        result = self.run_cli("install", "--replace")
+
+        self.assertNotEqual(0, result.returncode)
+        self.assertIn("сначала", result.stderr)
+        self.assertIn("stop", result.stderr)
+        self.assertEqual(state_before, self.managed_state_path().read_bytes())
+        self.assertEqual(xray_before, self.xray.read_bytes())
+        self.assertFalse((self.home / ".local/bin/xray").exists())
+        self.assert_no_secrets(result.stdout + result.stderr, profile_text)
+        self.stop_and_collect(process)
+
     def test_stale_cleanup_waits_for_start_lock_and_preserves_files(self):
         state_dir = self.state_home / "xray-proxy"
         state_dir.mkdir(parents=True, mode=0o700)