zapret-kvn/tests/test_app_update_check.py

from __future__ import annotations

import json
import unittest
from unittest.mock import patch
import urllib.error

import base64
import hashlib

from xray_fluent.updates import app_updater, signed_release
from xray_fluent.network.http_utils import HttpFetchError, HttpResponseData
from zapret_update import manifest as update_manifest
from zapret_update import mirrors as git_mirrors
from zapret_update.ed25519 import public_key_from_seed, sign_ed25519

SEED = hashlib.sha256(b"kvn test key").digest()
KEYS = {"kvn-test": public_key_from_seed(SEED)}


def _response(payload: bytes, url: str) -> HttpResponseData:
    return HttpResponseData(payload, url, 200, "proxy")


def _signed_release(digest, *, seed=SEED, listed_digest=None, with_list=True, listed_version="9.9.9", files=()):
    """Данные выпуска 9.9.9 и ответы сервера: выпуск, сумма архива, список, подпись."""
    asset_name = "ZapretKVN-v9.9.9-windows-x64.zip"
    base = "https://git2.zapret.moe/zapretkvn/zapret-kvn/releases/download/v9.9.9/"
    raw = update_manifest.dumps(
        {
            "schema": 1, "product": "zapret-kvn", "channel": "stable", "version": listed_version,
            "created": 1, "key": "kvn-test", "files": list(files),
            "installer": {"name": asset_name, "size": 10, "sha256": listed_digest or digest},
        }
    )
    assets = [
        {"name": asset_name, "browser_download_url": base + asset_name, "size": 10},
        {"name": asset_name + ".sha256", "browser_download_url": base + asset_name + ".sha256"},
    ]
    if with_list:
        for name in (update_manifest.MANIFEST_NAME, update_manifest.SIGNATURE_NAME):
            assets.append({"name": name, "browser_download_url": base + name})
    metadata = {"tag_name": "v9.9.9", "assets": assets}
    responses = [
        _response(json.dumps(metadata).encode(), app_updater.FORGEJO_RELEASE_API),
        _response((digest + "  " + asset_name).encode(), base + asset_name + ".sha256"),
    ]
    if with_list:
        responses += [
            _response(raw, base + update_manifest.MANIFEST_NAME),
            _response(base64.b64encode(sign_ed25519(seed, raw)), base + update_manifest.SIGNATURE_NAME),
        ]
    return metadata, responses


class AppUpdateCheckTests(unittest.TestCase):
    def setUp(self) -> None:
        # Список имён сервера в этих тестах не нужен: обмен идёт под основным именем.
        patcher = patch.object(git_mirrors.MirrorList, "hosts", return_value=["git.zapret.moe"])
        patcher.start()
        self.addCleanup(patcher.stop)

    def test_release_client_uses_active_proxy_and_bounded_retry_policy(self) -> None:
        payload = json.dumps({"tag_name": "v0.4.86", "assets": []}).encode()
        with patch.object(
            app_updater,
            "fetch_bytes",
            return_value=_response(payload, app_updater.FORGEJO_RELEASE_API),
        ) as fetch:
            result = app_updater._ReleaseClient(
                "http://127.0.0.1:1391"
            ).fetch_release()

        self.assertEqual(result["tag_name"], "v0.4.86")
        kwargs = fetch.call_args.kwargs
        self.assertEqual(kwargs["proxy_url"], "http://127.0.0.1:1391")
        self.assertEqual(kwargs["attempts_per_route"], 2)
        self.assertTrue(kwargs["prefer_proxy"])
        self.assertEqual(kwargs["max_bytes"], 1024 * 1024)
        self.assertEqual(kwargs["fallback_http_statuses"], frozenset({451}))

    def test_checksum_sidecar_uses_same_transport_and_size_limit(self) -> None:
        digest = "a" * 64
        metadata, responses = _signed_release(digest)

        with patch.object(signed_release, "TRUSTED_KEYS", KEYS), patch.object(
            app_updater, "fetch_bytes", side_effect=responses
        ) as fetch:
            update = app_updater._find_available_update("http://127.0.0.1:1391")

        self.assertIsNotNone(update)
        self.assertEqual(update.digest_sha256, digest)
        self.assertEqual(fetch.call_args_list[1].kwargs["max_bytes"], 16 * 1024)
        for call in fetch.call_args_list:
            self.assertEqual(call.kwargs["proxy_url"], "http://127.0.0.1:1391")

    def test_signed_file_list_travels_with_the_offered_update(self) -> None:
        listed = [{"path": "ZapretKVN.exe", "size": 3, "sha256": "c" * 64}]
        _metadata, responses = _signed_release("a" * 64, files=listed)
        with patch.object(signed_release, "TRUSTED_KEYS", KEYS), patch.object(
            app_updater, "fetch_bytes", side_effect=responses
        ):
            update = app_updater._find_available_update()
        self.assertEqual(
            [(item.path, item.size, item.sha256) for item in update.files], [("ZapretKVN.exe", 3, "c" * 64)]
        )

    def test_archive_is_offered_only_with_a_valid_signed_list(self) -> None:
        cases = {
            "чужая подпись": dict(seed=hashlib.sha256(b"stranger").digest()),
            "другая сумма у Forgejo": dict(listed_digest="b" * 64),
            "нет списка": dict(with_list=False),
            "старая версия в списке": dict(listed_version="0.0.1"),
        }
        for name, options in cases.items():
            with self.subTest(name):
                _metadata, responses = _signed_release("a" * 64, **options)
                with patch.object(signed_release, "TRUSTED_KEYS", KEYS), patch.object(
                    app_updater, "fetch_bytes", side_effect=responses
                ):
                    with self.assertRaises(app_updater.UpdateRejected):
                        app_updater._find_available_update()

    def test_untrusted_release_api_redirect_is_rejected(self) -> None:
        payload = json.dumps({"tag_name": "v0.4.86", "assets": []}).encode()
        with patch.object(
            app_updater,
            "fetch_bytes",
            return_value=_response(payload, "https://example.com/releases/latest"),
        ):
            with self.assertRaisesRegex(ValueError, "недоверенный"):
                app_updater._ReleaseClient().fetch_release()

    def test_network_exhaustion_has_friendly_error_without_raw_urllib(self) -> None:
        error = HttpFetchError((TimeoutError("handshake timed out"),))
        message = app_updater._describe_update_check_error(error, has_proxy=False)

        self.assertIn("временно не отвечает", message)
        self.assertNotIn("urllib", message)
        self.assertNotIn("handshake", message)

    def test_451_without_proxy_recommends_connecting(self) -> None:
        cause = urllib.error.HTTPError(
            app_updater.FORGEJO_RELEASE_API,
            451,
            "Unavailable For Legal Reasons",
            {},
            None,
        )
        message = app_updater._describe_update_check_error(
            HttpFetchError((cause,)),
            has_proxy=False,
        )

        self.assertIn("Подключитесь к серверу", message)
        self.assertNotIn("Unavailable", message)

    def test_update_checker_emits_friendly_network_error(self) -> None:
        checker = app_updater.UpdateChecker()
        emitted: list[str] = []
        checker.error.connect(emitted.append)

        with (
            patch.object(
                app_updater,
                "_find_available_update",
                side_effect=HttpFetchError((TimeoutError("handshake timed out"),)),
            ),
            patch.object(app_updater._log, "warning"),
        ):
            checker.run()

        self.assertEqual(len(emitted), 1)
        self.assertIn("временно не отвечает", emitted[0])
        self.assertNotIn("handshake", emitted[0])


if __name__ == "__main__":
    unittest.main()


class UpdateDownloaderTests(unittest.TestCase):
    """Загрузчик отдаёт установщику распакованную сборку или внятный отказ."""

    def _downloader(self, files: dict[str, str], *, digest: str | None = None):
        import hashlib
        import io
        import zipfile

        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, "w") as bundle:
            for name, text in files.items():
                bundle.writestr(name, text)
        payload = buffer.getvalue()
        update = app_updater.AppUpdate(
            version="9.9.9",
            tag="v9.9.9",
            download_url="https://git2.zapret.moe/zapretkvn/zapret-kvn/releases/download/v9.9.9/a.zip",
            size=len(payload),
            notes="",
            digest_sha256=hashlib.sha256(payload).hexdigest() if digest is None else digest,
        )
        downloader = app_updater.UpdateDownloader(update)
        downloader._download = lambda archive, proxy_url: archive.write_bytes(payload)
        self.errors: list[str] = []
        downloader.error.connect(self.errors.append)
        return downloader

    def setUp(self) -> None:
        import tempfile

        self._tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self._tmp.cleanup)
        patcher = patch.object(app_updater.tempfile, "gettempdir", return_value=self._tmp.name)
        patcher.start()
        self.addCleanup(patcher.stop)
        patcher = patch.object(
            app_updater.tempfile, "mkdtemp",
            side_effect=lambda prefix: self._make_work_dir(prefix),
        )
        patcher.start()
        self.addCleanup(patcher.stop)

    def _make_work_dir(self, prefix: str) -> str:
        from pathlib import Path

        path = Path(self._tmp.name) / f"{prefix}test"
        path.mkdir()
        return str(path)

    def test_valid_archive_becomes_a_prepared_update(self) -> None:
        downloader = self._downloader({"ZapretKVN/ZapretKVN.exe": "exe", "ZapretKVN/core/xray.exe": "x"})
        downloader.run()
        self.assertEqual(self.errors, [])
        prepared = downloader.prepared
        self.assertEqual(prepared.version, "9.9.9")
        self.assertTrue((prepared.source_dir / "ZapretKVN.exe").is_file())
        self.assertEqual(prepared.source_dir.name, "ZapretKVN")
        self.assertEqual(prepared.source_dir.parent.parent, prepared.work_dir)
        # Архив после распаковки не нужен.
        self.assertFalse((prepared.work_dir / "update.zip").exists())

    def test_wrong_checksum_is_a_permanent_failure_and_leaves_nothing_behind(self) -> None:
        downloader = self._downloader({"ZapretKVN/ZapretKVN.exe": "exe"}, digest="0" * 64)
        downloader.run()
        self.assertIsNone(downloader.prepared)
        self.assertTrue(downloader.failure_permanent)
        self.assertEqual(self.errors, ["Контрольная сумма архива не совпадает"])
        self.assertEqual(list(__import__("pathlib").Path(self._tmp.name).iterdir()), [])

    def test_archive_without_the_application_is_rejected(self) -> None:
        downloader = self._downloader({"readme.txt": "nothing here"})
        downloader.run()
        self.assertIsNone(downloader.prepared)
        self.assertTrue(downloader.failure_permanent)

    def test_network_failure_is_not_permanent(self) -> None:
        downloader = self._downloader({"ZapretKVN/ZapretKVN.exe": "exe"})

        def offline(archive, proxy_url):
            raise OSError("no route")

        downloader._download = offline
        downloader.run()
        self.assertIsNone(downloader.prepared)
        self.assertFalse(downloader.failure_permanent)
        self.assertEqual(len(self.errors), 1)


class ReleaseClientMirrorNamesTests(unittest.TestCase):
    """Основное имя режут по TLS: проверка обновления идёт под первым открытым именем сервера."""

    HOSTS = ["git2.zapret.moe", "git3.zapret.moe", "git.zapret.moe"]

    def _client(self):
        client = app_updater._ReleaseClient()
        client._mirrors._hosts = list(self.HOSTS)
        client._mirrors._expires_at = float("inf")
        return client

    @staticmethod
    def _api(host: str) -> str:
        return git_mirrors.with_host(app_updater.FORGEJO_RELEASE_API, host)

    def test_next_name_is_tried_only_after_every_route_failed(self) -> None:
        client = self._client()
        payload = json.dumps({"tag_name": "v1.0.0", "assets": []}).encode()
        calls: list[str] = []

        def fake(request, **_kwargs):
            calls.append(request.full_url)
            if "git2." in request.full_url:
                raise HttpFetchError(())
            return _response(payload, request.full_url)

        with patch.object(app_updater, "fetch_bytes", side_effect=fake):
            release = client.fetch_release()

        self.assertEqual(release["tag_name"], "v1.0.0")
        self.assertEqual(calls, [self._api("git2.zapret.moe"), self._api("git3.zapret.moe")])
        self.assertEqual(client._mirrors.hosts()[0], "git3.zapret.moe")

    def test_server_answer_like_404_is_not_retried_under_other_names(self) -> None:
        client = self._client()
        error = urllib.error.HTTPError(self._api("git2.zapret.moe"), 404, "нет", {}, None)

        with patch.object(app_updater, "fetch_bytes", side_effect=error) as fetch:
            with self.assertRaises(urllib.error.HTTPError):
                client.fetch_release()

        self.assertEqual(fetch.call_count, 1)

    def test_when_no_name_answers_the_network_error_is_raised(self) -> None:
        client = self._client()

        with patch.object(app_updater, "fetch_bytes", side_effect=HttpFetchError(())) as fetch:
            with self.assertRaises(HttpFetchError):
                client.fetch_release()

        self.assertEqual(fetch.call_count, 3)

    def test_answer_from_a_mirror_is_trusted_but_a_foreign_redirect_is_not(self) -> None:
        client = self._client()
        payload = json.dumps({"tag_name": "v1.0.0", "assets": []}).encode()

        with patch.object(app_updater, "fetch_bytes", return_value=_response(payload, self._api("git2.zapret.moe"))):
            self.assertEqual(client.fetch_release()["tag_name"], "v1.0.0")

        with patch.object(app_updater, "fetch_bytes", return_value=_response(payload, "https://evil.example/x")):
            with self.assertRaises(ValueError):
                client.fetch_release()

    def test_release_file_links_on_any_project_git_name_are_trusted(self) -> None:
        path = "/zapretkvn/zapret-kvn/releases/download/v1/a.zip"
        for host in ("git.zapret.moe", "git2.zapret.moe", "git9.zapret.moe"):
            self.assertTrue(app_updater._is_trusted_release_url(f"https://{host}{path}"), host)
        for host in ("evil.example", "git2.zapret.moe.evil.example", "wiki.zapret.moe"):
            self.assertFalse(app_updater._is_trusted_release_url(f"https://{host}{path}"), host)
        self.assertFalse(app_updater._is_trusted_release_url(f"http://git2.zapret.moe{path}"))
        self.assertFalse(app_updater._is_trusted_release_url(f"https://git2.zapret.moe:8443{path}"))