Initial SyncVito release with installer updater
This commit is contained in:
Davito committed
commit
b9e4c09f49
41 files changed
+4331
No files matched your search
@@ -0,0 +1,426 @@
|
||||
"""Sicherer Gitea-Updater für den installierten SyncVito-Client."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from collections.abc import Callable, Iterable
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from syncvito.config_store import default_data_directory
|
||||
|
||||
__all__ = [
|
||||
"GiteaUpdateService",
|
||||
"UpdateError",
|
||||
"UpdateFormatError",
|
||||
"UpdateInfo",
|
||||
"UpdateInstallError",
|
||||
"UpdateIntegrityError",
|
||||
"UpdateNetworkError",
|
||||
"parse_version",
|
||||
]
|
||||
|
||||
GITEA_BASE_URL = "https://gitea.davito.synology.me"
|
||||
GITEA_OWNER = "Davito"
|
||||
GITEA_REPOSITORY = "SyncVito"
|
||||
MAX_RELEASE_BYTES = 2 * 1024 * 1024
|
||||
MAX_CHECKSUM_BYTES = 1024 * 1024
|
||||
MAX_INSTALLER_BYTES = 1024 * 1024 * 1024
|
||||
USER_AGENT = "SyncVito-AutoUpdater"
|
||||
ProgressCallback = Callable[[int, str], None]
|
||||
_VERSION_PATTERN = re.compile(r"^v?(\d+)\.(\d+)\.(\d+)$")
|
||||
_SHA256_PATTERN = re.compile(r"^[0-9a-fA-F]{64}$")
|
||||
_MANAGED_INSTALLER_PATTERN = re.compile(
|
||||
r"^SyncVitoSetup-\d+\.\d+\.\d+\.exe(?:\.part)?$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
class UpdateError(RuntimeError):
|
||||
"""Basisklasse für erwartete Updatefehler."""
|
||||
|
||||
|
||||
class UpdateNetworkError(UpdateError):
|
||||
"""Gitea oder ein Release-Artefakt war nicht erreichbar."""
|
||||
|
||||
|
||||
class UpdateFormatError(UpdateError):
|
||||
"""Release-Metadaten waren unvollständig oder ungültig."""
|
||||
|
||||
|
||||
class UpdateIntegrityError(UpdateError):
|
||||
"""Die heruntergeladene Datei bestand die Integritätsprüfung nicht."""
|
||||
|
||||
|
||||
class UpdateInstallError(UpdateError):
|
||||
"""Der externe Installer-Wächter konnte nicht gestartet werden."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class UpdateInfo:
|
||||
"""Validierte Metadaten eines installierbaren Gitea-Releases."""
|
||||
|
||||
version: str
|
||||
title: str
|
||||
notes: str
|
||||
release_url: str
|
||||
installer_name: str
|
||||
download_url: str
|
||||
checksum_url: str
|
||||
size: int
|
||||
|
||||
|
||||
def parse_version(value: str) -> tuple[int, int, int]:
|
||||
"""Konvertiert stabile SemVer-Tags wie ``v1.2.3`` in Vergleichswerte."""
|
||||
match = _VERSION_PATTERN.fullmatch(str(value or "").strip())
|
||||
if match is None:
|
||||
raise ValueError(f"Ungültige stabile Versionsnummer: {value}")
|
||||
return tuple(int(part) for part in match.groups()) # type: ignore[return-value]
|
||||
|
||||
|
||||
def _powershell_literal(value: str | Path) -> str:
|
||||
return "'" + str(value).replace("'", "''") + "'"
|
||||
|
||||
|
||||
def _start_detached(executable: str, arguments: list[str]) -> None:
|
||||
"""Startet einen Prozess unabhängig vom Lebenszyklus der Anwendung."""
|
||||
if os.name == "nt":
|
||||
os.startfile(
|
||||
executable,
|
||||
operation="open",
|
||||
arguments=subprocess.list2cmdline(arguments),
|
||||
show_cmd=0,
|
||||
)
|
||||
return
|
||||
subprocess.Popen(
|
||||
[executable, *arguments],
|
||||
stdin=subprocess.DEVNULL,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
close_fds=True,
|
||||
)
|
||||
|
||||
|
||||
def _start_installer_after_exit(
|
||||
installer_path: Path,
|
||||
arguments: list[str],
|
||||
wait_pids: Iterable[int],
|
||||
bootstrap_log_path: Path,
|
||||
) -> None:
|
||||
"""Startet Inno Setup erst nach dem Ende aller PyInstaller-Prozesse."""
|
||||
if os.name != "nt":
|
||||
_start_detached(str(installer_path), arguments)
|
||||
return
|
||||
|
||||
system_root = os.environ.get("SYSTEMROOT", r"C:\Windows")
|
||||
powershell_path = Path(system_root) / "System32" / "WindowsPowerShell" / "v1.0" / "powershell.exe"
|
||||
if not powershell_path.is_file():
|
||||
raise UpdateInstallError(f"Windows PowerShell wurde nicht gefunden: {powershell_path}")
|
||||
|
||||
normalized_pids = sorted({int(pid) for pid in wait_pids if int(pid) > 0})
|
||||
pid_list = ", ".join(str(pid) for pid in normalized_pids)
|
||||
setup_arguments = ", ".join(_powershell_literal(argument) for argument in arguments)
|
||||
script = f"""
|
||||
$ErrorActionPreference = 'Stop'
|
||||
$bootstrapLog = {_powershell_literal(bootstrap_log_path)}
|
||||
try {{
|
||||
[System.IO.File]::WriteAllText(
|
||||
$bootstrapLog,
|
||||
((Get-Date).ToString('o') + ' Warte auf SyncVito-Prozesse: {pid_list}' + [Environment]::NewLine)
|
||||
)
|
||||
foreach ($processId in @({pid_list})) {{
|
||||
$deadline = [DateTime]::UtcNow.AddMinutes(2)
|
||||
while (Get-Process -Id $processId -ErrorAction SilentlyContinue) {{
|
||||
if ([DateTime]::UtcNow -ge $deadline) {{
|
||||
throw "SyncVito-Prozess $processId wurde nicht innerhalb von zwei Minuten beendet."
|
||||
}}
|
||||
Start-Sleep -Milliseconds 200
|
||||
}}
|
||||
}}
|
||||
[System.IO.File]::AppendAllText(
|
||||
$bootstrapLog,
|
||||
((Get-Date).ToString('o') + ' Starte verifizierten Installer.' + [Environment]::NewLine)
|
||||
)
|
||||
$setupArguments = @({setup_arguments})
|
||||
& {_powershell_literal(installer_path)} @setupArguments
|
||||
$setupExitCode = $LASTEXITCODE
|
||||
[System.IO.File]::AppendAllText(
|
||||
$bootstrapLog,
|
||||
((Get-Date).ToString('o') + ' Installer-Exitcode: ' + $setupExitCode + [Environment]::NewLine)
|
||||
)
|
||||
exit $setupExitCode
|
||||
}}
|
||||
catch {{
|
||||
[System.IO.File]::AppendAllText(
|
||||
$bootstrapLog,
|
||||
((Get-Date).ToString('o') + ' Bootstrap-Fehler: ' + $_.Exception.Message + [Environment]::NewLine)
|
||||
)
|
||||
exit 1
|
||||
}}
|
||||
"""
|
||||
encoded_script = base64.b64encode(script.encode("utf-16-le")).decode("ascii")
|
||||
_start_detached(
|
||||
str(powershell_path),
|
||||
[
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
"-WindowStyle",
|
||||
"Hidden",
|
||||
"-EncodedCommand",
|
||||
encoded_script,
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
class _TrustedRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
def __init__(self, trusted_host: str) -> None:
|
||||
super().__init__()
|
||||
self.trusted_host = trusted_host
|
||||
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[no-untyped-def]
|
||||
_validate_update_url(newurl, self.trusted_host)
|
||||
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
||||
|
||||
|
||||
def _validate_update_url(url: str, trusted_host: str) -> str:
|
||||
parsed = urllib.parse.urlparse(str(url or ""))
|
||||
try:
|
||||
port = parsed.port
|
||||
except ValueError as exc:
|
||||
raise UpdateFormatError("Gitea lieferte eine ungültige Downloadadresse.") from exc
|
||||
if (
|
||||
parsed.scheme.lower() != "https"
|
||||
or parsed.hostname != trusted_host
|
||||
or port not in (None, 443)
|
||||
or parsed.username is not None
|
||||
or parsed.password is not None
|
||||
):
|
||||
raise UpdateFormatError("Gitea lieferte eine nicht vertrauenswürdige Downloadadresse.")
|
||||
return parsed.geturl()
|
||||
|
||||
|
||||
class GiteaUpdateService:
|
||||
"""Kapselt das öffentliche Installer-Updateprotokoll für SyncVito."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
current_version: str,
|
||||
base_url: str = GITEA_BASE_URL,
|
||||
owner: str = GITEA_OWNER,
|
||||
repository: str = GITEA_REPOSITORY,
|
||||
update_directory: Path | None = None,
|
||||
) -> None:
|
||||
self.current_version = current_version
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.owner = owner
|
||||
self.repository = repository
|
||||
self.update_directory = update_directory or default_data_directory() / "updates"
|
||||
parsed_base = urllib.parse.urlparse(self.base_url)
|
||||
if parsed_base.scheme.lower() != "https" or not parsed_base.hostname:
|
||||
raise ValueError("Die Updatequelle muss eine gültige HTTPS-Adresse sein.")
|
||||
self.trusted_host = parsed_base.hostname
|
||||
parse_version(current_version)
|
||||
|
||||
def check(self) -> UpdateInfo | None:
|
||||
"""Liefert den neuesten stabilen Installer, sofern er neuer ist."""
|
||||
endpoint = f"{self.base_url}/api/v1/repos/{self.owner}/{self.repository}/releases/latest"
|
||||
try:
|
||||
payload = self._read_json(endpoint)
|
||||
except UpdateNetworkError as exc:
|
||||
if "HTTP 404" in str(exc):
|
||||
return None
|
||||
raise
|
||||
if not isinstance(payload, dict) or payload.get("draft") or payload.get("prerelease"):
|
||||
raise UpdateFormatError("Gitea lieferte keinen stabilen Release.")
|
||||
try:
|
||||
version_tuple = parse_version(str(payload.get("tag_name", "")))
|
||||
except ValueError as exc:
|
||||
raise UpdateFormatError("Der neueste Gitea-Release besitzt keine gültige Version.") from exc
|
||||
if version_tuple <= parse_version(self.current_version):
|
||||
return None
|
||||
version = ".".join(str(part) for part in version_tuple)
|
||||
return self._parse_release(payload, version)
|
||||
|
||||
def download(self, info: UpdateInfo, progress: ProgressCallback | None = None) -> Path:
|
||||
"""Lädt den Installer und veröffentlicht ihn erst nach SHA-256-Prüfung."""
|
||||
self.update_directory.mkdir(parents=True, exist_ok=True)
|
||||
expected_hash = self._read_checksum(info.checksum_url, info.installer_name)
|
||||
destination = self.update_directory / info.installer_name
|
||||
temporary = destination.with_suffix(".exe.part")
|
||||
digest = hashlib.sha256()
|
||||
received = 0
|
||||
try:
|
||||
with self._request(info.download_url, timeout=30) as response, temporary.open("wb") as output:
|
||||
announced = int(response.headers.get("Content-Length", info.size or 0))
|
||||
if announced > MAX_INSTALLER_BYTES:
|
||||
raise UpdateFormatError("Der angebotene Installer ist unerwartet groß.")
|
||||
while chunk := response.read(1024 * 1024):
|
||||
received += len(chunk)
|
||||
if received > MAX_INSTALLER_BYTES:
|
||||
raise UpdateFormatError("Der geladene Installer ist unerwartet groß.")
|
||||
output.write(chunk)
|
||||
digest.update(chunk)
|
||||
if progress and announced > 0:
|
||||
progress(min(99, round(received / announced * 100)), "Installer wird heruntergeladen …")
|
||||
if received <= 0:
|
||||
raise UpdateFormatError("Der heruntergeladene Installer ist leer.")
|
||||
if digest.hexdigest().lower() != expected_hash:
|
||||
raise UpdateIntegrityError(
|
||||
"Die SHA-256-Prüfung des Installers ist fehlgeschlagen; die Datei wird nicht ausgeführt."
|
||||
)
|
||||
os.replace(temporary, destination)
|
||||
except UpdateError:
|
||||
temporary.unlink(missing_ok=True)
|
||||
raise
|
||||
except (OSError, TimeoutError, urllib.error.URLError, ValueError) as exc:
|
||||
temporary.unlink(missing_ok=True)
|
||||
raise UpdateNetworkError(f"Der Installer konnte nicht heruntergeladen werden: {exc}") from exc
|
||||
if progress:
|
||||
progress(100, "Installer geprüft und bereit")
|
||||
self._remove_old_downloads(destination)
|
||||
return destination
|
||||
|
||||
def launch_installer(self, installer_path: Path, version: str) -> None:
|
||||
"""Startet einen Wächter, der nach dem App-Ende Inno Setup ausführt."""
|
||||
if os.name != "nt" or not getattr(sys, "frozen", False):
|
||||
raise UpdateInstallError("Updates können nur aus der installierten Windows-Version gestartet werden.")
|
||||
installer = installer_path.resolve()
|
||||
expected_name = f"SyncVitoSetup-{version}.exe"
|
||||
if installer.parent != self.update_directory.resolve() or installer.name != expected_name or not installer.is_file():
|
||||
raise UpdateInstallError("Der geprüfte Installer wurde nicht gefunden.")
|
||||
|
||||
safe_version = re.sub(r"[^0-9A-Za-z._-]+", "_", version).strip("._") or "unknown"
|
||||
setup_log = installer.with_name(f"SyncVitoSetup-{safe_version}.log")
|
||||
bootstrap_log = installer.with_name(f"SyncVitoUpdater-{safe_version}-bootstrap.log")
|
||||
current_executable = Path(sys.executable).resolve()
|
||||
wait_pids = [os.getpid()]
|
||||
parent_pid = os.getppid()
|
||||
if parent_pid > 0 and parent_pid not in wait_pids:
|
||||
wait_pids.append(parent_pid)
|
||||
arguments = [
|
||||
"/VERYSILENT",
|
||||
"/SUPPRESSMSGBOXES",
|
||||
"/NORESTART",
|
||||
"/CLOSEAPPLICATIONS",
|
||||
f"/DIR={current_executable.parent}",
|
||||
f"/LOG={setup_log}",
|
||||
]
|
||||
try:
|
||||
_start_installer_after_exit(installer, arguments, wait_pids, bootstrap_log)
|
||||
except (OSError, ValueError, UpdateError) as exc:
|
||||
raise UpdateInstallError(f"Die Installation konnte nicht gestartet werden: {exc}") from exc
|
||||
|
||||
def cleanup_installers(self) -> tuple[int, int]:
|
||||
"""Entfernt ausschließlich von SyncVito verwaltete Setup-Dateien."""
|
||||
try:
|
||||
candidates = tuple(self.update_directory.resolve().iterdir())
|
||||
except FileNotFoundError:
|
||||
return 0, 0
|
||||
except OSError:
|
||||
return 0, 1
|
||||
removed = 0
|
||||
pending = 0
|
||||
for candidate in candidates:
|
||||
if not _MANAGED_INSTALLER_PATTERN.fullmatch(candidate.name) or candidate.is_dir():
|
||||
continue
|
||||
try:
|
||||
candidate.unlink()
|
||||
except OSError:
|
||||
pending += 1
|
||||
else:
|
||||
removed += 1
|
||||
return removed, pending
|
||||
|
||||
def _parse_release(self, release: dict[str, Any], version: str) -> UpdateInfo:
|
||||
assets = release.get("assets")
|
||||
if not isinstance(assets, list):
|
||||
raise UpdateFormatError(f"Release {version} enthält keine Dateianhänge.")
|
||||
by_name = {str(asset.get("name")): asset for asset in assets if isinstance(asset, dict)}
|
||||
installer_name = f"SyncVitoSetup-{version}.exe"
|
||||
checksum_name = f"{installer_name}.sha256"
|
||||
installer = by_name.get(installer_name)
|
||||
checksum = by_name.get(checksum_name)
|
||||
if installer is None or checksum is None:
|
||||
raise UpdateFormatError(f"Release {version} benötigt {installer_name} und {checksum_name}.")
|
||||
download_url = _validate_update_url(str(installer.get("browser_download_url", "")), self.trusted_host)
|
||||
checksum_url = _validate_update_url(str(checksum.get("browser_download_url", "")), self.trusted_host)
|
||||
return UpdateInfo(
|
||||
version=version,
|
||||
title=str(release.get("name") or f"SyncVito {version}"),
|
||||
notes=str(release.get("body") or "Keine Versionshinweise vorhanden."),
|
||||
release_url=str(release.get("html_url") or self.base_url),
|
||||
installer_name=installer_name,
|
||||
download_url=download_url,
|
||||
checksum_url=checksum_url,
|
||||
size=max(0, int(installer.get("size") or 0)),
|
||||
)
|
||||
|
||||
def _request(self, url: str, timeout: int): # type: ignore[no-untyped-def]
|
||||
request = urllib.request.Request(
|
||||
_validate_update_url(url, self.trusted_host),
|
||||
headers={"Accept": "application/json", "User-Agent": USER_AGENT},
|
||||
)
|
||||
opener = urllib.request.build_opener(_TrustedRedirectHandler(self.trusted_host))
|
||||
try:
|
||||
response = opener.open(request, timeout=timeout)
|
||||
_validate_update_url(response.geturl(), self.trusted_host)
|
||||
return response
|
||||
except urllib.error.HTTPError as exc:
|
||||
raise UpdateNetworkError(f"Gitea antwortete mit HTTP {exc.code}.") from exc
|
||||
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
||||
raise UpdateNetworkError(f"Der Gitea-Server ist nicht erreichbar: {exc}") from exc
|
||||
|
||||
def _read_json(self, url: str) -> Any:
|
||||
try:
|
||||
with self._request(url, timeout=12) as response:
|
||||
payload = response.read(MAX_RELEASE_BYTES + 1)
|
||||
if len(payload) > MAX_RELEASE_BYTES:
|
||||
raise UpdateFormatError("Die Release-Antwort ist unerwartet groß.")
|
||||
return json.loads(payload)
|
||||
except UpdateError:
|
||||
raise
|
||||
except (json.JSONDecodeError, UnicodeDecodeError, TypeError) as exc:
|
||||
raise UpdateFormatError("Gitea hat ungültige Release-Daten geliefert.") from exc
|
||||
|
||||
def _read_checksum(self, url: str, filename: str) -> str:
|
||||
try:
|
||||
with self._request(url, timeout=12) as response:
|
||||
payload = response.read(MAX_CHECKSUM_BYTES + 1)
|
||||
if len(payload) > MAX_CHECKSUM_BYTES:
|
||||
raise UpdateFormatError("Die Prüfsummendatei ist unerwartet groß.")
|
||||
content = payload.decode("utf-8")
|
||||
except UpdateError:
|
||||
raise
|
||||
except UnicodeDecodeError as exc:
|
||||
raise UpdateFormatError("Die Update-Prüfsumme ist ungültig.") from exc
|
||||
fallback = ""
|
||||
for raw_line in content.splitlines():
|
||||
parts = raw_line.strip().replace(" ", " ").split()
|
||||
if not parts or not _SHA256_PATTERN.fullmatch(parts[0]):
|
||||
continue
|
||||
digest = parts[0].lower()
|
||||
if len(parts) == 1:
|
||||
fallback = digest
|
||||
elif " ".join(parts[1:]).lstrip("*").strip().casefold() == filename.casefold():
|
||||
return digest
|
||||
if fallback:
|
||||
return fallback
|
||||
raise UpdateFormatError(f"Keine SHA-256-Prüfsumme für {filename} gefunden.")
|
||||
|
||||
def _remove_old_downloads(self, keep: Path) -> None:
|
||||
for candidate in self.update_directory.glob("SyncVitoSetup-*.exe"):
|
||||
if candidate != keep:
|
||||
try:
|
||||
candidate.unlink()
|
||||
except OSError:
|
||||
continue
|
||||
Reference in new issue
Block a user