WGSecure/app/core/updater.py

"""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.")