WGSecure/app/utils/platform_utils.py

import sys
import os
import subprocess
import platform


def is_windows() -> bool:
    return sys.platform == "win32"


def is_linux() -> bool:
    return sys.platform.startswith("linux")


def get_config_dir() -> str:
    if is_windows():
        base = os.environ.get("APPDATA", os.path.expanduser("~"))
        path = os.path.join(base, "WGSecure")
    else:
        path = os.path.join(os.path.expanduser("~"), ".wgsecure")
    os.makedirs(path, exist_ok=True)
    return path


def get_wg_config_dir() -> str:
    if is_windows():
        base = os.environ.get("PROGRAMDATA", r"C:\ProgramData")
        return os.path.join(base, "WireGuard")
    return "/etc/wireguard"


def _windows_wg_dir() -> str:
    """Répertoire d'installation de WireGuard for Windows."""
    for base in (os.environ.get("ProgramFiles", r"C:\Program Files"),
                 os.environ.get("ProgramW6432", r"C:\Program Files"),
                 os.environ.get("ProgramFiles(x86)", "")):
        if not base:
            continue
        path = os.path.join(base, "WireGuard")
        if os.path.isdir(path):
            return path
    return ""


def _windows_exe(name: str) -> str:
    """Chemin absolu de wireguard.exe / wg.exe.

    L'installateur WireGuard n'ajoute pas son répertoire au PATH : invoquer
    « wireguard » ou « wg » tel quel remonte « Commande introuvable » sur une
    machine où WireGuard est pourtant installé.
    """
    import shutil as _sh
    d = _windows_wg_dir()
    if d:
        path = os.path.join(d, name)
        if os.path.isfile(path):
            return path
    return _sh.which(name) or name


def wireguard_exe() -> str:
    """Binaire du gestionnaire WireGuard (installe/désinstalle les tunnels)."""
    return _windows_exe("wireguard.exe") if is_windows() else "wg-quick"


def wg_exe() -> str:
    """Binaire `wg` (interrogation d'un tunnel actif)."""
    return _windows_exe("wg.exe") if is_windows() else "wg"


def has_root_privileges() -> bool:
    if is_windows():
        try:
            import ctypes
            return ctypes.windll.shell32.IsUserAnAdmin() != 0
        except Exception:
            return False
    return os.geteuid() == 0


# Windows ouvre une fenêtre console pour chaque processus console lancé depuis
# une application GUI. WGSecure interroge `ping` et `wireguard` en boucle pour
# rafraîchir le statut : sans ce drapeau, l'utilisateur voit des fenêtres DOS
# apparaître et disparaître en permanence. CREATE_NO_WINDOW supprime la
# console ; startupinfo/SW_HIDE couvre les lanceurs qui l'ignorent.
_CREATE_NO_WINDOW = 0x08000000


def _no_window_kwargs() -> dict:
    """Options subprocess empêchant l'apparition d'une console sous Windows."""
    if not is_windows():
        return {}
    startupinfo = subprocess.STARTUPINFO()
    startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
    startupinfo.wShowWindow = subprocess.SW_HIDE
    return {
        "creationflags": _CREATE_NO_WINDOW,
        "startupinfo": startupinfo,
    }


def run_command(cmd: list[str], timeout: int = 10,
                env: dict | None = None) -> tuple[int, str, str]:
    try:
        text_kwargs = {"text": True}
        if is_windows():
            # Les outils console natifs (sc.exe, PowerShell non-Core, ping…)
            # écrivent dans le codepage OEM de la console (ex. CP850 en
            # français), pas en UTF-8 : sans préciser l'encodage, Python
            # décode avec le codepage ANSI par défaut et tout caractère
            # accentué remonte corrompu ("échoué" → "‚chou‚").
            text_kwargs = {"encoding": "oem", "errors": "replace"}
        result = subprocess.run(
            cmd,
            capture_output=True,
            timeout=timeout,
            env=env,
            **text_kwargs,
            **_no_window_kwargs(),
        )
        return result.returncode, result.stdout.strip(), result.stderr.strip()
    except subprocess.TimeoutExpired:
        return -1, "", "Timeout"
    except FileNotFoundError:
        return -1, "", f"Commande introuvable : {cmd[0]}"
    except Exception as e:
        return -1, "", str(e)


def _c_locale_env() -> dict:
    """Environnement forçant les messages système en anglais.

    Sans cela, la détection « sudo demande un mot de passe » échoue sur une
    session en français (« mot de passe » au lieu de « password »), et l'appel
    repart avec le code d'erreur de sudo au lieu de basculer sur pkexec.
    """
    env = os.environ.copy()
    env["LC_ALL"] = "C"
    env["LANG"] = "C"
    return env


# Messages que sudo émet quand il refuse *avant* de lancer la commande.
# Comparés en anglais : _c_locale_env() force LC_ALL=C pour que cette
# détection ne dépende pas de la langue de la session.
_SUDO_AUTH_MARKERS = (
    "a password is required",
    "no askpass program specified",
    "no tty present",
    "must have a tty",
    "a terminal is required",
    "is not allowed to run",
    "may not run sudo",
    "authentication failure",
)


def _sudo_refused_auth(err: str) -> bool:
    """True si sudo a refusé faute d'authentification, sans lancer la commande.

    Distinguer ce cas d'un échec de la commande elle-même est indispensable :
    dans le premier, il faut basculer sur askpass/pkexec ; dans le second, il
    faut remonter l'erreur telle quelle à l'utilisateur.

    Interroger `sudo -l` ne convient pas : pour un compte membre du groupe
    sudo (`(ALL : ALL) ALL`), `sudo -n -l -- <cmd>` répond « autorisé » pour
    *toutes* les commandes, y compris celles qui exigeront un mot de passe.
    """
    low = err.lower()
    return any(marker in low for marker in _SUDO_AUTH_MARKERS)


def _ps_quote(s: str) -> str:
    """Échappe une chaîne pour une chaîne litérale PowerShell entre quotes simples."""
    return "'" + s.replace("'", "''") + "'"


def _run_elevated_windows(cmd: list[str], timeout: int = 60) -> tuple[int, str, str]:
    """Relance `cmd` derrière une invite UAC et attend sa fin.

    `wireguard.exe /installtunnelservice` et `wireguard.exe /uninstalltunnelservice`
    exigent les droits administrateur : lancés depuis une session standard, ils
    échouaient sur « Access is denied » sans que l'application le sache. ShellExecuteEx
    avec le verbe « runas » est le seul moyen d'obtenir l'élévation depuis un
    processus déjà démarré.

    Un processus élevé (démarré par le courtier UAC, hors relation parent/
    enfant) n'hérite jamais de nos tubes : stdout/stderr ne peuvent pas
    remonter par héritage, avec ou sans redirection côté appelant (même
    `Start-Process -Verb RunAs -RedirectStandardOutput` échoue pour la même
    raison). Le seul moyen fiable : élever `powershell.exe` et lui faire
    écrire *lui-même*, une fois élevé, la sortie de `cmd` dans un fichier
    temporaire, puis relire ce fichier une fois le process terminé.
    """
    import ctypes
    from ctypes import wintypes
    import tempfile
    import uuid

    SEE_MASK_NOCLOSEPROCESS = 0x00000040
    SEE_MASK_NOASYNC = 0x00000100
    SW_HIDE = 0
    ERROR_CANCELLED = 1223
    WAIT_TIMEOUT = 0x00000102

    class SHELLEXECUTEINFOW(ctypes.Structure):
        _fields_ = [
            ("cbSize", wintypes.DWORD),
            ("fMask", ctypes.c_ulong),
            ("hwnd", wintypes.HWND),
            ("lpVerb", wintypes.LPCWSTR),
            ("lpFile", wintypes.LPCWSTR),
            ("lpParameters", wintypes.LPCWSTR),
            ("lpDirectory", wintypes.LPCWSTR),
            ("nShow", ctypes.c_int),
            ("hInstApp", wintypes.HINSTANCE),
            ("lpIDList", ctypes.c_void_p),
            ("lpClass", wintypes.LPCWSTR),
            ("hkeyClass", wintypes.HKEY),
            ("dwHotKey", wintypes.DWORD),
            ("hIconOrMonitor", wintypes.HANDLE),
            ("hProcess", wintypes.HANDLE),
        ]

    out_path = os.path.join(tempfile.gettempdir(), f"wgsecure-elev-{uuid.uuid4().hex}.log")
    # `*>&1 | Out-File -Encoding utf8` plutôt que la redirection `*>` brute :
    # cette dernière écrit en UTF-16LE (encodage par défaut de PowerShell
    # Windows pour `>`/`*>`), que la lecture Python en UTF-8 corromprait
    # (accents illisibles). `*>&1` fusionne tous les flux (succès, erreur,
    # avertissement…) dans le flux de sortie avant le pipe, comme le faisait
    # `*>` seul — cohérent avec l'usage existant qui traite `err or out`
    # indifféremment.
    call_expr = "& " + " ".join(_ps_quote(a) for a in cmd)
    wrapped_script = (
        f"{call_expr} *>&1 | Out-File -FilePath {_ps_quote(out_path)} -Encoding utf8; "
        f"exit $LASTEXITCODE"
    )
    elevated_cmd = ["powershell", "-NoProfile", "-NonInteractive", "-Command", wrapped_script]

    info = SHELLEXECUTEINFOW()
    info.cbSize = ctypes.sizeof(info)
    info.fMask = SEE_MASK_NOCLOSEPROCESS | SEE_MASK_NOASYNC
    info.lpVerb = "runas"
    info.lpFile = elevated_cmd[0]
    info.lpParameters = subprocess.list2cmdline(elevated_cmd[1:])
    info.nShow = SW_HIDE

    if not ctypes.windll.shell32.ShellExecuteExW(ctypes.byref(info)):
        err = ctypes.get_last_error() or ctypes.GetLastError()
        if err == ERROR_CANCELLED:
            return -1, "", "Élévation refusée par l'utilisateur"
        return -1, "", f"Élévation impossible (code {err}) : {cmd[0]}"

    handle = info.hProcess

    def _read_output() -> str:
        try:
            # utf-8-sig : Out-File -Encoding utf8 (Windows PowerShell)
            # ajoute un BOM ; -sig l'ignore silencieusement s'il est présent.
            with open(out_path, encoding="utf-8-sig", errors="replace") as f:
                return f.read().strip()
        except OSError:
            return ""
        finally:
            try:
                os.unlink(out_path)
            except OSError:
                pass

    if not handle:
        # Commande lancée sans handle exploitable : succès non vérifiable.
        return 0, _read_output(), ""
    try:
        if ctypes.windll.kernel32.WaitForSingleObject(
            handle, int(timeout * 1000)
        ) == WAIT_TIMEOUT:
            return -1, "", "Timeout"
        code = wintypes.DWORD()
        ctypes.windll.kernel32.GetExitCodeProcess(handle, ctypes.byref(code))
        return code.value, _read_output(), ""
    finally:
        ctypes.windll.kernel32.CloseHandle(handle)


def run_privileged(cmd: list[str], timeout: int = 60) -> tuple[int, str, str]:
    if is_windows():
        if has_root_privileges():
            return run_command(cmd, timeout)
        try:
            return _run_elevated_windows(cmd, timeout)
        except Exception as e:
            return -1, "", str(e)
    if has_root_privileges():
        return run_command(cmd, timeout)

    # 1. sudo -n : réussit sans aucun dialogue si une règle NOPASSWD couvre la
    #    commande (cf. `make setup-sudoers`).
    code, out, err = run_command(["sudo", "-n"] + cmd, timeout,
                                 env=_c_locale_env())
    if code == 0:
        return code, out, err
    if not _sudo_refused_auth(err):
        # sudo a bien lancé la commande : c'est elle qui a échoué. Inutile de
        # redemander des privilèges, l'erreur est remontée telle quelle.
        return code, out, err

    # 2. sudo avec programme askpass graphique (pas de TTY dans une app Qt)
    askpass = _find_askpass()
    if askpass:
        env = _c_locale_env()
        env["SUDO_ASKPASS"] = askpass
        code, out, err = run_command(["sudo", "-A"] + cmd, timeout, env=env)
        if code == 0:
            return code, out, err

    # 3. pkexec — dialogue graphique polkit (GNOME/KDE)
    code, out, err = run_command(["pkexec"] + cmd, timeout, env=_c_locale_env())
    # 126 = authentification refusée/annulée, 127 = pkexec absent : on préfère
    # le message d'aide au code opaque.
    if code not in (-1, 126, 127):
        return code, out, err

    return (
        -1, "",
        "Élévation de privilèges impossible.\n"
        "Exécutez  make setup-sudoers  pour autoriser WGSecure sans dialogue.\n"
        "Ou manuellement :\n"
        f"  sudo bash -c \"echo '{os.environ.get('USER','<user>')} ALL=(ALL) NOPASSWD: /usr/bin/wg-quick'"
        " > /etc/sudoers.d/wgsecure && chmod 440 /etc/sudoers.d/wgsecure\""
    )


def run_privileged_readonly(cmd: list[str], timeout: int = 10) -> tuple[int, str, str]:
    """Comme `run_privileged`, mais pour un sondage périodique en lecture seule.

    N'ouvre jamais de dialogue (askpass/pkexec/UAC) : un sondage lancé toutes
    les quelques secondes (état du tunnel, handshake) rendrait l'application
    inutilisable si chaque appel pouvait faire apparaître une invite. Ne
    fonctionne donc que si le process est déjà root/administrateur, ou si la
    commande est couverte par une règle sudo NOPASSWD (`make setup-sudoers`).
    Échoue silencieusement sinon, comme un manque de privilèges ordinaire.
    """
    if has_root_privileges():
        return run_command(cmd, timeout)
    if is_windows():
        # `run_command` n'élève jamais (pas de fenêtre UAC) : la tentative
        # est donc sans risque, et peut réussir si l'utilisateur a le
        # mécanisme LimitedOperatorUI + groupe « Network Configuration
        # Operators » (cf. _grant_wireguard_permissions dans admin_window.py).
        # Un abandon systématique sans essayer condamnait le handshake à
        # rester vide même quand l'accès non élevé fonctionne réellement.
        return run_command(cmd, timeout)
    return run_command(["sudo", "-n"] + cmd, timeout, env=_c_locale_env())


def _find_askpass() -> str:
    """Cherche un programme askpass graphique disponible."""
    candidates = [
        os.environ.get("SUDO_ASKPASS", ""),
        "/usr/lib/openssh/gnome-ssh-askpass",
        "/usr/bin/ssh-askpass",
        "/usr/bin/x11-ssh-askpass",
        "/usr/libexec/ssh-askpass",
    ]
    for p in candidates:
        if p and os.path.isfile(p) and os.access(p, os.X_OK):
            return p
    # Cherche aussi via shutil
    import shutil
    for name in ("ssh-askpass", "x11-ssh-askpass", "gnome-ssh-askpass"):
        found = shutil.which(name)
        if found:
            return found
    return ""


# Commandes que WGSecure lance avec des privilèges, et ce qu'elles servent.
#
# `make setup-sudoers` installe des règles NOPASSWD restreintes à des motifs
# d'arguments précis (ex. `ip link delete dev *`), jamais à la commande nue.
# Les sondes doivent donc reproduire un appel réel qui matche ce motif :
# tester avec `--version`/`-V` interroge un appel que sudoers n'autorise
# jamais, et affiche « dialogue à chaque appel » même quand tout est
# correctement configuré.
_PRIVILEGE_PROBE_IFACE = "wgsecure-privcheck0"


def _install_probe_cmd() -> list[str]:
    """Reproduit le motif `install -D -m 600 -o root -g root * /etc/wireguard/*`.

    Écrit un fichier vide et sans conséquence (`.wgsecure-privcheck`, jamais
    lu par wg-quick) : c'est le seul moyen fiable de tester ce motif, `sudo
    -n -l` ne le pouvant pas (cf. `_sudo_refused_auth`).
    """
    dest = os.path.join(get_wg_config_dir(), ".wgsecure-privcheck")
    return ["install", "-D", "-m", "600", "-o", "root", "-g", "root",
            "/dev/null", dest]


def _privileged_commands() -> tuple[tuple[list[str], str], ...]:
    return (
        (["wg-quick", "--help"], "monter / démonter le tunnel"),
        (_install_probe_cmd(), "écrire la config dans /etc/wireguard"),
        (["ip", "link", "delete", "dev", _PRIVILEGE_PROBE_IFACE],
         "supprimer une interface orpheline"),
        (["resolvconf", "-d", _PRIVILEGE_PROBE_IFACE],
         "retirer l'entrée DNS du tunnel"),
        (["resolvectl", "revert", _PRIVILEGE_PROBE_IFACE],
         "purger le DNS systemd-resolved"),
        (["resolvectl", "dns", _PRIVILEGE_PROBE_IFACE, "0.0.0.0"],
         "configurer le DNS distant (split-DNS)"),
        (["resolvectl", "domain", _PRIVILEGE_PROBE_IFACE, "~invalid"],
         "configurer le domaine distant (split-DNS)"),
    )


def privilege_report() -> list[tuple[str, bool, str]]:
    """État des droits pour chaque commande privilégiée utilisée.

    Retourne [(commande, sans_mot_de_passe, rôle)]. Une commande absente du
    système est ignorée. Sert à expliquer pourquoi un dialogue polkit apparaît
    à chaque connexion plutôt que de laisser l'utilisateur deviner.
    """
    import shutil as _sh
    report: list[tuple[str, bool, str]] = []
    if is_windows() or has_root_privileges():
        return report
    for probe, role in _privileged_commands():
        if not _sh.which(probe[0]):
            continue
        code, _, err = run_command(["sudo", "-n"] + probe, 8, env=_c_locale_env())
        report.append((probe[0], not _sudo_refused_auth(err), role))
    return report


def wg_available() -> bool:
    if is_windows():
        return os.path.isfile(wg_exe())
    code, _, _ = run_command(["wg", "--version"])
    return code == 0


def wg_quick_available() -> bool:
    # wireguard.exe est une application graphique : elle n'accepte pas
    # « --help » et n'écrit rien sur stdout. Le sonder comme un outil console
    # concluait « WireGuard absent » sur toute machine Windows. Sa présence sur
    # le disque est le seul test fiable.
    if is_windows():
        return os.path.isfile(wireguard_exe())
    code, _, _ = run_command(["wg-quick", "--help"])
    return code == 0


def launch_command() -> str:
    """Commande à écrire dans .desktop / registre Run pour relancer l'appli.

    Dans un binaire PyInstaller, `main.py` n'est pas déployé : écrire
    « python3 <chemin>/main.py » produisait un lanceur et un démarrage
    automatique qui ne fonctionnaient jamais depuis un paquet installé.
    """
    if getattr(sys, "frozen", False):
        return f'"{sys.executable}"' if " " in sys.executable else sys.executable
    script = os.path.abspath(
        os.path.join(os.path.dirname(__file__), "..", "..", "main.py")
    )
    exe = "pythonw" if is_windows() else "python3"
    return f'{exe} "{script}"' if " " in script else f"{exe} {script}"


def install_desktop_entry(icon_png_path: str) -> None:
    """Installe l'entrée .desktop et l'icône PNG pour la barre des tâches GNOME/KDE.
    Appelé une fois au démarrage ; sans effet sur Windows."""
    if is_windows():
        return
    try:
        # Icône dans le thème hicolor
        icon_dir = os.path.expanduser("~/.local/share/icons/hicolor/256x256/apps")
        os.makedirs(icon_dir, exist_ok=True)
        dest_icon = os.path.join(icon_dir, "wgsecure.png")
        import shutil
        shutil.copy2(icon_png_path, dest_icon)

        # Fichier .desktop
        app_dir = os.path.expanduser("~/.local/share/applications")
        os.makedirs(app_dir, exist_ok=True)
        desktop = os.path.join(app_dir, "wgsecure.desktop")
        content = (
            "[Desktop Entry]\n"
            "Type=Application\n"
            "Name=WGSecure\n"
            "Comment=WireGuard GUI avec MFA TOTP\n"
            f"Exec={launch_command()}\n"
            "Icon=wgsecure\n"
            "Categories=Network;Security;\n"
            "StartupWMClass=main\n"
            "Terminal=false\n"
        )
        with open(desktop, "w") as f:
            f.write(content)

        # Mettre à jour le cache d'icônes si disponible
        run_command(["gtk-update-icon-cache", "-f", "-t",
                     os.path.expanduser("~/.local/share/icons/hicolor")])
    except Exception:
        pass