Danil Iarosh
f8f6a216f5
- Добавлена проверка и безопасная установка OTA-обновлений Windows с откатом. - Нумерация линий и блоков переведена на индексы с нуля. - Добавлена миграция старых JSON-схем с сохранением физической адресации. - Обновлены генератор, сборка, документация и smoke-проверки. - Подготовлена версия 3.3.0.
327 lines
14 KiB
Python
327 lines
14 KiB
Python
"""Безопасная проверка и установка Windows-релизов LED Mapper."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import platform
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import BinaryIO, Callable
|
||
from urllib.parse import urlparse
|
||
from urllib.request import Request, urlopen
|
||
|
||
try:
|
||
from .version import __version__
|
||
except ImportError:
|
||
from version import __version__
|
||
|
||
RELEASE_API_URL = "https://git.robotail.pro/api/v1/repos/Robotail/LED-mapper/releases/latest"
|
||
RELEASE_PAGE_URL = "https://git.robotail.pro/Robotail/LED-mapper/releases"
|
||
RELEASE_HOST = "git.robotail.pro"
|
||
READY_ENV = "LEDMAPPER_UPDATE_READY_FILE"
|
||
MAX_METADATA_BYTES = 1024 * 1024
|
||
MAX_DOWNLOAD_BYTES = 200 * 1024 * 1024
|
||
CHUNK_BYTES = 1024 * 1024
|
||
VERSION_RE = re.compile(r"v?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)\Z")
|
||
EXECUTABLE_RE = re.compile(r"LEDMapper-(v\d+\.\d+\.\d+)\.exe\Z", re.IGNORECASE)
|
||
|
||
|
||
class UpdateError(RuntimeError):
|
||
"""Проверка, загрузка или установка обновления не выполнена."""
|
||
|
||
|
||
class UpdateCancelled(UpdateError):
|
||
"""Загрузка отменена пользователем."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class UpdateInfo:
|
||
tag: str
|
||
version: str
|
||
asset_name: str
|
||
download_url: str
|
||
size: int
|
||
|
||
|
||
def parse_version(value: str) -> tuple[int, int, int]:
|
||
match = VERSION_RE.fullmatch(value.strip())
|
||
if match is None:
|
||
raise UpdateError("Версия релиза должна иметь формат vX.Y.Z")
|
||
return tuple(int(part) for part in match.groups())
|
||
|
||
|
||
def current_version() -> str:
|
||
if getattr(sys, "frozen", False):
|
||
bundle_dir = getattr(sys, "_MEIPASS", None)
|
||
if bundle_dir is not None:
|
||
try:
|
||
bundled_version = (Path(bundle_dir) / "ledmapper-build-version.txt").read_text(encoding="utf-8").strip()
|
||
if VERSION_RE.fullmatch(bundled_version):
|
||
return bundled_version
|
||
except OSError:
|
||
pass
|
||
match = EXECUTABLE_RE.fullmatch(Path(sys.executable).name)
|
||
if match is not None:
|
||
return match.group(1)
|
||
return __version__
|
||
|
||
|
||
def _trusted_url(value: str) -> bool:
|
||
parsed = urlparse(value)
|
||
return (
|
||
parsed.scheme == "https"
|
||
and parsed.hostname == RELEASE_HOST
|
||
and parsed.port in (None, 443)
|
||
and not parsed.username
|
||
and not parsed.password
|
||
and not parsed.fragment
|
||
)
|
||
|
||
|
||
def check_for_update(
|
||
installed_version: str,
|
||
*,
|
||
opener: Callable[..., BinaryIO] | None = None,
|
||
timeout: float = 10.0,
|
||
) -> UpdateInfo | None:
|
||
"""Возвращает только новый релиз с ровно одним подходящим Windows EXE."""
|
||
installed = parse_version(installed_version)
|
||
if platform.system() != "Windows":
|
||
raise UpdateError("Автообновление доступно только в Windows")
|
||
request = Request(
|
||
RELEASE_API_URL,
|
||
headers={"Accept": "application/json", "User-Agent": "LEDMapper-Updater"},
|
||
)
|
||
try:
|
||
with (opener or urlopen)(request, timeout=timeout) as response:
|
||
if not _trusted_url(response.geturl()):
|
||
raise UpdateError("Сервер релизов перенаправил запрос на другой сайт")
|
||
payload = response.read(MAX_METADATA_BYTES + 1)
|
||
if len(payload) > MAX_METADATA_BYTES:
|
||
raise UpdateError("Ответ сервера релизов слишком велик")
|
||
release = json.loads(payload)
|
||
except UpdateError:
|
||
raise
|
||
except (OSError, ValueError, TypeError) as exc:
|
||
raise UpdateError(f"Не удалось проверить обновления: {exc}") from exc
|
||
if not isinstance(release, dict) or release.get("draft") or release.get("prerelease"):
|
||
raise UpdateError("Сервер не вернул опубликованный стабильный релиз")
|
||
tag = release.get("tag_name")
|
||
if not isinstance(tag, str) or not tag.startswith("v"):
|
||
raise UpdateError("У релиза отсутствует корректный тег")
|
||
remote = parse_version(tag)
|
||
if remote <= installed:
|
||
return None
|
||
expected_name = f"LEDMapper-{tag}.exe"
|
||
assets = release.get("assets")
|
||
if not isinstance(assets, list):
|
||
raise UpdateError("У релиза отсутствует список артефактов")
|
||
matching = [asset for asset in assets if isinstance(asset, dict) and asset.get("name") == expected_name]
|
||
if len(matching) != 1:
|
||
raise UpdateError(f"Не найден Windows-артефакт {expected_name}")
|
||
asset = matching[0]
|
||
download_url = asset.get("browser_download_url")
|
||
expected_path = f"/Robotail/LED-mapper/releases/download/{tag}/{expected_name}"
|
||
if not isinstance(download_url, str) or not _trusted_url(download_url) or urlparse(download_url).path != expected_path:
|
||
raise UpdateError("Релиз содержит неожиданный адрес артефакта")
|
||
size = asset.get("size")
|
||
if isinstance(size, bool) or not isinstance(size, int) or not 0 < size <= MAX_DOWNLOAD_BYTES:
|
||
raise UpdateError("Релиз содержит недопустимый размер артефакта")
|
||
return UpdateInfo(tag, tag.removeprefix("v"), expected_name, download_url, size)
|
||
|
||
|
||
def _check_executable(path: Path) -> None:
|
||
with path.open("rb") as source:
|
||
header = source.read(64)
|
||
if len(header) < 64 or header[:2] != b"MZ":
|
||
raise UpdateError("Скачанный файл не является Windows EXE")
|
||
pe_offset = int.from_bytes(header[0x3C:0x40], "little")
|
||
if not 64 <= pe_offset <= 1024 * 1024:
|
||
raise UpdateError("Некорректный заголовок Windows EXE")
|
||
source.seek(pe_offset)
|
||
if source.read(4) != b"PE\x00\x00":
|
||
raise UpdateError("Некорректный заголовок Windows EXE")
|
||
if source.read(2) != b"\x64\x86":
|
||
raise UpdateError("Обновление не предназначено для Windows x64")
|
||
|
||
|
||
def download_update(
|
||
info: UpdateInfo,
|
||
directory: Path,
|
||
*,
|
||
opener: Callable[..., BinaryIO] | None = None,
|
||
progress: Callable[[int], None] | None = None,
|
||
cancelled: Callable[[], bool] | None = None,
|
||
timeout: float = 30.0,
|
||
) -> Path:
|
||
"""Скачивает артефакт в отдельный временный файл и проверяет размер/PE."""
|
||
target_dir = directory.resolve()
|
||
if not target_dir.is_dir():
|
||
raise UpdateError("Каталог приложения не найден")
|
||
if not _trusted_url(info.download_url) or urlparse(info.download_url).path != (
|
||
f"/Robotail/LED-mapper/releases/download/{info.tag}/{info.asset_name}"
|
||
):
|
||
raise UpdateError("Недопустимый адрес загрузки обновления")
|
||
path: Path | None = None
|
||
received = 0
|
||
request = Request(info.download_url, headers={"User-Agent": "LEDMapper-Updater"})
|
||
try:
|
||
with (opener or urlopen)(request, timeout=timeout) as response:
|
||
if not _trusted_url(response.geturl()):
|
||
raise UpdateError("Загрузка перенаправлена на другой сайт")
|
||
with tempfile.NamedTemporaryFile(
|
||
mode="wb", dir=target_dir, prefix=".ledmapper-update-", suffix=".download", delete=False
|
||
) as output:
|
||
path = Path(output.name)
|
||
while True:
|
||
if cancelled is not None and cancelled():
|
||
raise UpdateCancelled("Загрузка обновления отменена")
|
||
chunk = response.read(CHUNK_BYTES)
|
||
if not chunk:
|
||
break
|
||
received += len(chunk)
|
||
if received > MAX_DOWNLOAD_BYTES or received > info.size:
|
||
raise UpdateError("Артефакт превышает объявленный размер")
|
||
output.write(chunk)
|
||
if progress is not None:
|
||
progress(min(received * 100 // info.size, 100))
|
||
output.flush()
|
||
os.fsync(output.fileno())
|
||
if received != info.size:
|
||
raise UpdateError("Артефакт скачан не полностью")
|
||
_check_executable(path)
|
||
if progress is not None:
|
||
progress(100)
|
||
return path
|
||
except (UpdateError, OSError) as exc:
|
||
if path is not None:
|
||
path.unlink(missing_ok=True)
|
||
if isinstance(exc, UpdateError):
|
||
raise
|
||
raise UpdateError(f"Не удалось скачать обновление: {exc}") from exc
|
||
|
||
|
||
def current_executable() -> Path:
|
||
if not getattr(sys, "frozen", False):
|
||
raise UpdateError("Установка обновления доступна только в собранном EXE")
|
||
return Path(sys.executable).resolve()
|
||
|
||
|
||
def signal_update_ready() -> None:
|
||
marker = os.environ.pop(READY_ENV, None)
|
||
if marker is None:
|
||
return
|
||
try:
|
||
path = Path(marker).resolve()
|
||
if path.parent != current_executable().parent or not path.name.startswith(".ledmapper-ready-"):
|
||
return
|
||
path.write_text(str(os.getpid()), encoding="ascii")
|
||
except (OSError, UpdateError):
|
||
pass
|
||
|
||
|
||
def _write_windows_updater(path: Path) -> None:
|
||
script = r'''param(
|
||
[int]$ApplicationProcessId,
|
||
[string]$Source,
|
||
[string]$OldTarget,
|
||
[string]$NewTarget,
|
||
[string]$Backup,
|
||
[string]$ReadyFile
|
||
)
|
||
$ErrorActionPreference = "Stop"
|
||
$applicationDirectory = Split-Path -Parent $OldTarget
|
||
$errorLog = Join-Path $applicationDirectory "update-error.log"
|
||
$restarted = $null
|
||
$oldMoved = $false
|
||
$newMoved = $false
|
||
try {
|
||
Wait-Process -Id $ApplicationProcessId -ErrorAction SilentlyContinue
|
||
# Родительский процесс PyInstaller может ещё удерживать EXE при очистке.
|
||
$moveDeadline = (Get-Date).AddSeconds(30)
|
||
while (-not $oldMoved) {
|
||
try {
|
||
Move-Item -LiteralPath $OldTarget -Destination $Backup -ErrorAction Stop
|
||
$oldMoved = $true
|
||
} catch {
|
||
if ((Get-Date) -ge $moveDeadline) { throw }
|
||
Start-Sleep -Milliseconds 250
|
||
}
|
||
}
|
||
if (Test-Path -LiteralPath $NewTarget) { throw "Целевой EXE уже существует" }
|
||
Move-Item -LiteralPath $Source -Destination $NewTarget -ErrorAction Stop
|
||
$newMoved = $true
|
||
$env:LEDMAPPER_UPDATE_READY_FILE = $ReadyFile
|
||
$env:PYINSTALLER_RESET_ENVIRONMENT = "1"
|
||
$restarted = Start-Process -FilePath $NewTarget -WorkingDirectory $applicationDirectory -PassThru
|
||
$deadline = (Get-Date).AddSeconds(60)
|
||
while (-not (Test-Path -LiteralPath $ReadyFile -PathType Leaf)) {
|
||
if ($restarted.HasExited) { throw "Новая версия завершилась до открытия окна" }
|
||
if ((Get-Date) -ge $deadline) { throw "Новая версия не подтвердила запуск" }
|
||
Start-Sleep -Milliseconds 250
|
||
}
|
||
if ($restarted.HasExited) { throw "Новая версия завершилась сразу после запуска" }
|
||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||
Remove-Item -LiteralPath $Backup -Force -ErrorAction SilentlyContinue
|
||
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force -ErrorAction SilentlyContinue
|
||
} catch {
|
||
$failure = $_.ToString()
|
||
if ($restarted -and -not $restarted.HasExited) {
|
||
& taskkill.exe /PID $restarted.Id /T /F *> $null
|
||
}
|
||
if ($newMoved -and (Test-Path -LiteralPath $NewTarget)) {
|
||
Remove-Item -LiteralPath $NewTarget -Force -ErrorAction SilentlyContinue
|
||
}
|
||
if ($oldMoved -and (Test-Path -LiteralPath $Backup)) {
|
||
Move-Item -LiteralPath $Backup -Destination $OldTarget -Force -ErrorAction SilentlyContinue
|
||
}
|
||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||
$failure | Out-File -LiteralPath $errorLog -Encoding utf8
|
||
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force -ErrorAction SilentlyContinue
|
||
exit 1
|
||
}
|
||
'''
|
||
path.write_text(script, encoding="utf-8-sig", newline="\r\n")
|
||
|
||
|
||
def launch_update(downloaded: Path, info: UpdateInfo, *, process_id: int | None = None) -> None:
|
||
"""Запускает отдельный helper; замену EXE он делает после выхода приложения."""
|
||
old_target = current_executable()
|
||
source = downloaded.resolve()
|
||
if platform.system() != "Windows":
|
||
raise UpdateError("Установка обновления доступна только в Windows")
|
||
if source.parent != old_target.parent or not source.is_file() or not source.name.startswith(".ledmapper-update-"):
|
||
raise UpdateError("Скачанный файл должен находиться рядом с приложением")
|
||
if not EXECUTABLE_RE.fullmatch(info.asset_name):
|
||
raise UpdateError("Недопустимое имя Windows-артефакта")
|
||
new_target = old_target
|
||
if source.stat().st_size != info.size:
|
||
raise UpdateError("Размер скачанного артефакта изменился")
|
||
_check_executable(source)
|
||
token = uuid.uuid4().hex
|
||
script_path = old_target.parent / f".ledmapper-updater-{token}.ps1"
|
||
backup_path = old_target.parent / f".ledmapper-backup-{token}.exe"
|
||
ready_path = old_target.parent / f".ledmapper-ready-{token}"
|
||
try:
|
||
_write_windows_updater(script_path)
|
||
subprocess.Popen(
|
||
[
|
||
"powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File",
|
||
str(script_path), str(process_id or os.getpid()), str(source), str(old_target),
|
||
str(new_target), str(backup_path), str(ready_path),
|
||
],
|
||
close_fds=True,
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||
)
|
||
except (OSError, UpdateError) as exc:
|
||
script_path.unlink(missing_ok=True)
|
||
if isinstance(exc, UpdateError):
|
||
raise
|
||
raise UpdateError(f"Не удалось запустить установщик: {exc}") from exc
|