428 lines
17 KiB
Python
428 lines
17 KiB
Python
"""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)
|
|
)
|
|
$env:PYINSTALLER_RESET_ENVIRONMENT = '1'
|
|
$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
|