"""Mise à jour du client, servie par le serveur depuis l'intérieur du tunnel. Les binaires ne peuvent pas venir du dépôt Gitea : il est privé, et un jeton embarqué dans une application distribuée est un jeton donné à tous ses utilisateurs. Ils viennent donc du serveur WGSecure, où l'accès est déjà tranché — l'API n'est joignable que par le tunnel, et on y parvient parce que sa clé WireGuard est déclarée. Aucun secret à embarquer ici. Rien ne s'installe sans un clic : la vérification est automatique, la suite ne l'est pas. Cette application monte un VPN, et une version défaillante qui se propagerait seule couperait l'accès de tout un parc sans que personne ne l'ait demandé. L'archive téléchargée est vérifiée contre le condensat SHA-256 annoncé avant d'être installée. Le canal est déjà authentifié par WireGuard ; le condensat couvre ce que le chiffrement ne couvre pas — un fichier tronqué, un disque plein, un artefact mal publié. """ from __future__ import annotations import hashlib import json import os import sys import tempfile import urllib.error import urllib.request from typing import Callable, NamedTuple from app.core.config import Config from app.core.vpn_session import _OPENER, api_base_url from app.utils.helper_ipc import call_helper from app.utils.platform_utils import is_windows # Le sondage part au montage du tunnel, dans un thread que l'arrêt de # l'application attend : bref, donc, comme celui de l'état d'accès. _CHECK_TIMEOUT = 5 # Le téléchargement, lui, porte sur 125 Mo : la borne couvre une liaison lente, # pas un serveur muet — l'absence de premier octet est tranchée par le sondage. _DOWNLOAD_TIMEOUT = 600 _CHUNK = 256 * 1024 class Update(NamedTuple): """Une version disponible et l'artefact qui convient à ce poste.""" version: str notes: str key: str filename: str size: int sha256: str def _parse_version(raw: str) -> tuple[int, ...]: """« 0.10.0 » → (0, 10, 0), pour comparer autrement qu'alphabétiquement. Comparées comme des chaînes, « 0.9.1 » passait pour plus récent que « 0.10.0 » : le neuvième caractère décidait de tout. """ parts = [] for chunk in (raw or "").strip().lstrip("vV").split("."): digits = "".join(c for c in chunk if c.isdigit()) parts.append(int(digits) if digits else 0) return tuple(parts) or (0,) def artifact_key() -> str: """Artefact attendu par ce poste. Sous Windows, c'est l'installeur et non l'exécutable nu : le `.exe` en cours ne peut pas être remplacé pendant qu'il tourne, alors que l'installeur sait arrêter, remplacer et relancer. """ return "windows-setup" if is_windows() else "linux-x86_64" def check(cfg: Config, current: str) -> Update | None: """Une version plus récente que `current` est-elle publiée ? `None` couvre tous les cas où il n'y a rien à proposer : serveur muet, serveur qui ne distribue pas de mise à jour, version identique ou plus ancienne, artefact absent pour cette plateforme. Aucun n'est une panne, et aucun ne doit se voir — une vérification qui échoue ne doit jamais interrompre quelqu'un qui voulait juste ouvrir son VPN. """ base = api_base_url(cfg) if not base: return None request = urllib.request.Request(f"{base}/api/version", method="GET") try: with _OPENER.open(request, timeout=_CHECK_TIMEOUT) as response: body = json.loads(response.read().decode("utf-8")) except (urllib.error.URLError, OSError, ValueError): return None latest = (body.get("version") or "").strip() if not latest or _parse_version(latest) <= _parse_version(current): return None key = artifact_key() entry = next((a for a in body.get("artifacts", []) if a.get("key") == key), None) if not entry or not entry.get("sha256"): return None return Update(version=latest, notes=body.get("notes", ""), key=key, filename=entry.get("filename", key), size=int(entry.get("size") or 0), sha256=entry["sha256"].lower()) def download(cfg: Config, update: Update, progress: Callable[[int, int], None] | None = None, dest_dir: str = "") -> str: """Récupère l'artefact et rend son chemin, ou lève. Écrit dans un fichier temporaire vérifié avant d'être présenté : un téléchargement interrompu ne doit jamais laisser derrière lui quelque chose qui ressemble à une mise à jour valide. `dest_dir` place ce temporaire là où le fichier finira. Il allait dans `/tmp`, souvent un autre système de fichiers que `~/.local/bin` : la mise en place se faisait alors par copie — donc par écriture sur l'exécutable en cours — et le noyau la refusait (« Fichier texte occupé »). Sur le même système de fichiers, c'est un simple renommage, que rien n'interdit. """ base = api_base_url(cfg) if not base: raise RuntimeError("Adresse du serveur de mise à jour inconnue.") request = urllib.request.Request( f"{base}/api/version/download/{update.key}", method="GET") digest = hashlib.sha256() received = 0 fd, tmp = tempfile.mkstemp(prefix=".wgsecure-update-", suffix=os.path.splitext(update.filename)[1], dir=dest_dir or None) try: with _OPENER.open(request, timeout=_DOWNLOAD_TIMEOUT) as response, \ os.fdopen(fd, "wb") as out: while True: chunk = response.read(_CHUNK) if not chunk: break out.write(chunk) digest.update(chunk) received += len(chunk) if progress: progress(received, update.size) except BaseException: os.unlink(tmp) raise if digest.hexdigest() != update.sha256: os.unlink(tmp) raise RuntimeError( "Le fichier téléchargé ne correspond pas à l'empreinte annoncée " "par le serveur : il est incomplet ou altéré. Rien n'a été " "installé.") if update.size and received != update.size: os.unlink(tmp) raise RuntimeError(f"Taille inattendue : {received} octets reçus pour " f"{update.size} annoncés.") return tmp def current_binary() -> str: """Chemin de l'exécutable à remplacer, "" hors binaire figé. Lancée depuis les sources (`python main.py`), l'application n'a pas de binaire à remplacer : la mise à jour ne s'y applique pas, et le proposer remplacerait l'interpréteur Python du système. """ return sys.executable if getattr(sys, "frozen", False) else "" def apply_update(archive: str) -> str: """Installe l'artefact téléchargé. Rend le message à montrer, ou lève. Deux systèmes, deux gestes. Sous Windows, l'installeur est lancé et l'application doit quitter : un `.exe` en cours d'exécution est verrouillé par le système, et se remplacer soi-même y est impossible. Sous Linux, l'ancien binaire est simplement délié puis le nouveau mis à sa place — le noyau garde le fichier ouvert vivant jusqu'à la fermeture du programme, qui continue donc de tourner normalement jusqu'à son redémarrage. """ if is_windows(): raise RuntimeError("Sous Windows, la mise à jour passe par " "`apply_via_helper`, pas par ce chemin.") target = current_binary() if not target: raise RuntimeError( "Mise à jour impossible : l'application tourne depuis les sources " "et non depuis un binaire installé.") try: os.chmod(archive, 0o755) # `os.replace` et non `shutil.move` : le premier est un renommage pur, # qui échoue plutôt que de se rabattre sur une copie. C'est ce repli # qui cassait la mise à jour — copier revient à écrire dans le fichier # exécuté, ce que Linux refuse (ETXTBSY). Le renommage, lui, ne touche # qu'une entrée de répertoire : l'ancien inode reste vivant pour le # processus en cours, qui continue de tourner jusqu'à sa fermeture. os.replace(archive, target) except OSError as exc: try: os.unlink(archive) except OSError: pass raise RuntimeError( f"Remplacement de {target} impossible : {exc}") from exc return ("Mise à jour installée. Elle prendra effet au prochain démarrage " "de WGSecure.") def apply_via_helper(cfg: Config) -> str: """Fait installer la mise à jour par le service, sans élévation. Le service tourne en LocalSystem : il écrit dans Program Files sans qu'une invite UAC n'apparaisse. Rien ne lui est transmis d'autre que l'adresse de l'API — ni fichier, ni chemin, ni empreinte. Il lit le manifeste, télécharge et vérifie lui-même. Lui passer un binaire déjà téléchargé serait plus rapide et ouvrirait à tout compte du poste le moyen de faire exécuter n'importe quoi avec les privilèges du système : le pipe est ouvert aux utilisateurs interactifs. Le service peut être absent — l'installeur le dépose sans l'enregistrer. L'appelant reçoit alors de quoi le dire, plutôt qu'un échec opaque. """ base = api_base_url(cfg) if not base: raise RuntimeError("Adresse du serveur de mise à jour inconnue.") reponse = call_helper({"cmd": "apply_update", "api_base": base}, timeout_ms=900_000) if reponse is None: raise RuntimeError( "Le service WGSecure Helper ne répond pas. Sans lui, la mise à " "jour demande les droits administrateur : lancez l'installeur " "téléchargé depuis le serveur.") if not reponse.get("ok"): raise RuntimeError(str(reponse.get("message") or "Mise à jour refusée.")) return str(reponse.get("message") or "Mise à jour installée.")