BUH_bot/modules/release_updater.py

"""Check public Gitea releases and securely stage the Windows installer."""

import hashlib
import os
import re
import tempfile
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlsplit

import requests
from packaging.version import InvalidVersion, Version

from app_version import (
    APP_VERSION,
    CHECKSUM_ASSET_NAME,
    GITEA_REPOSITORY_URL,
    INSTALLER_ASSET_NAME,
)


RELEASE_API_URL = (
    "https://git.robotail.pro/api/v1/repos/Robotail/BUH_bot/releases/latest"
)
REQUEST_TIMEOUT = (5, 30)


class UpdateError(Exception):
    """The release cannot be used to update this application."""


@dataclass(frozen=True)
class ReleaseInfo:
    tag_name: str
    version: Version
    notes: str
    page_url: str
    installer_url: str
    checksum_url: str
    installer_size: int


def _trusted_url(url: str) -> str:
    """Allow downloads only from the repository's HTTPS host."""
    try:
        parsed = urlsplit(url)
        repository_host = urlsplit(GITEA_REPOSITORY_URL).hostname
        trusted = (
            parsed.scheme == "https"
            and parsed.hostname == repository_host
            and parsed.port in (None, 443)
            and parsed.username is None
            and parsed.password is None
        )
    except (TypeError, ValueError):
        trusted = False
    if not trusted:
        raise UpdateError("Релиз содержит ссылку на неизвестный сервер")
    return url


def _version_from_tag(tag_name: str) -> Version:
    try:
        return Version(tag_name.removeprefix("v"))
    except InvalidVersion as error:
        raise UpdateError(f"Некорректный номер релиза: {tag_name}") from error


def get_latest_release(session=requests) -> ReleaseInfo | None:
    """Return the latest published stable Gitea release, if one exists."""
    try:
        response = session.get(RELEASE_API_URL, timeout=REQUEST_TIMEOUT)
        if response.status_code == 404:
            return None
        response.raise_for_status()
        data = response.json()
    except (requests.RequestException, ValueError) as error:
        raise UpdateError(f"Не удалось получить список релизов: {error}") from error

    if not isinstance(data, dict):
        raise UpdateError("Gitea вернула некорректные данные релиза")

    tag_name = data.get("tag_name")
    assets = data.get("assets")
    if not isinstance(tag_name, str) or not isinstance(assets, list):
        raise UpdateError("В релизе нет номера версии или списка файлов")

    installer = next(
        (asset for asset in assets if isinstance(asset, dict) and asset.get("name") == INSTALLER_ASSET_NAME),
        None,
    )
    checksum = next(
        (asset for asset in assets if isinstance(asset, dict) and asset.get("name") == CHECKSUM_ASSET_NAME),
        None,
    )
    if installer is None or checksum is None:
        raise UpdateError("В релизе нет установщика или файла контрольной суммы")

    try:
        installer_url = _trusted_url(installer["browser_download_url"])
        checksum_url = _trusted_url(checksum["browser_download_url"])
        page_url = _trusted_url(data["html_url"])
        installer_size = int(installer["size"])
    except (KeyError, TypeError, ValueError) as error:
        raise UpdateError("В релизе указаны некорректные файлы") from error
    if installer_size <= 0:
        raise UpdateError("В релизе указан пустой установщик")

    return ReleaseInfo(
        tag_name=tag_name,
        version=_version_from_tag(tag_name),
        notes=str(data.get("body") or ""),
        page_url=page_url,
        installer_url=installer_url,
        checksum_url=checksum_url,
        installer_size=installer_size,
    )


def update_available(release: ReleaseInfo) -> bool:
    return release.version > Version(APP_VERSION)


def _expected_checksum(release: ReleaseInfo, session) -> str:
    try:
        response = session.get(release.checksum_url, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()
        _trusted_url(response.url)
    except requests.RequestException as error:
        raise UpdateError(f"Не удалось получить контрольную сумму: {error}") from error

    checksum_line = response.text.strip()
    match = re.fullmatch(
        rf"([0-9a-fA-F]{{64}})(?:\s+\*?{re.escape(INSTALLER_ASSET_NAME)})?",
        checksum_line,
    )
    if not match:
        raise UpdateError("Файл контрольной суммы имеет неверный формат")
    return match.group(1).lower()


def download_installer(release: ReleaseInfo, session=requests) -> Path:
    """Download and verify the installer before it can be launched."""
    expected_hash = _expected_checksum(release, session)
    target_dir = Path(tempfile.gettempdir()) / "MoySkladGenerator" / release.tag_name
    target_dir.mkdir(parents=True, exist_ok=True)
    target = target_dir / INSTALLER_ASSET_NAME

    if target.is_file() and hashlib.sha256(target.read_bytes()).hexdigest() == expected_hash:
        return target

    temporary_path = None
    try:
        with session.get(release.installer_url, stream=True, timeout=REQUEST_TIMEOUT) as response:
            response.raise_for_status()
            _trusted_url(response.url)
            with tempfile.NamedTemporaryFile(
                dir=target_dir, prefix="installer-", suffix=".part", delete=False
            ) as temporary:
                temporary_path = Path(temporary.name)
                digest = hashlib.sha256()
                downloaded = 0
                for chunk in response.iter_content(chunk_size=1024 * 1024):
                    if chunk:
                        temporary.write(chunk)
                        digest.update(chunk)
                        downloaded += len(chunk)

        if downloaded != release.installer_size or digest.hexdigest() != expected_hash:
            raise UpdateError("Загруженный установщик повреждён или не прошёл проверку SHA-256")
        os.replace(temporary_path, target)
        return target
    except requests.RequestException as error:
        raise UpdateError(f"Не удалось загрузить установщик: {error}") from error
    finally:
        if temporary_path is not None:
            temporary_path.unlink(missing_ok=True)