WGSecure/app/utils/helper_ipc.py

"""Client IPC pour wgsecure-helper, le service Windows privilégié.

Utilise directement l'API Win32 via `ctypes` (même famille d'appels que
`_run_elevated_windows` dans `platform_utils.py`) plutôt que `pywin32`,
pour ne pas ajouter de dépendance à l'app principale — seul l'exécutable
séparé du service a besoin de `pywin32` (enregistrement SCM via
`win32serviceutil`), jamais `wgsecure.exe`.

Le service tourne en LocalSystem et applique, sans jamais demander
d'élévation UAC, les opérations que `platform_utils.run_privileged()`
demande sinon à chaque appel : installer/désinstaller le tunnel, poser/
retirer une règle NRPT (split-DNS), lire l'état du tunnel (`wg show`).
"""
from __future__ import annotations

import json
import sys

_PIPE_NAME = r"\\.\pipe\WGSecureHelper"


def call_helper(request: dict, timeout_ms: int = 1500) -> dict | None:
    """Envoie `request` (JSON) au service et attend sa réponse JSON.

    Retourne `None` si le service est absent, indisponible, ou si l'échange
    échoue pour n'importe quelle raison (pipe introuvable, occupé au-delà du
    timeout, réponse malformée…) — l'appelant doit alors basculer sur son
    chemin existant (élévation UAC). Ne lève jamais d'exception : cette
    fonction est un pur best-effort, jamais un point de défaillance.
    """
    if sys.platform != "win32":
        return None

    import ctypes
    from ctypes import wintypes

    handle = None
    kernel32 = None
    try:
        GENERIC_READ = 0x80000000
        GENERIC_WRITE = 0x40000000
        OPEN_EXISTING = 3
        FILE_ATTRIBUTE_NORMAL = 0x80
        INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value

        kernel32 = ctypes.windll.kernel32
        # Signatures explicites pour CHAQUE fonction utilisée : sans
        # argtypes, ctypes marshalle un HANDLE (pointeur 64 bits) comme un
        # c_int par défaut sur Windows 64 bits, le tronquant silencieusement
        # — ReadFile/WriteFile/CloseHandle échouaient alors sur un handle
        # corrompu, cet échec étant avalé par le retour `None` (best-
        # effort), ce qui faisait retomber *systématiquement* sur le chemin
        # élevé (UAC) sans que rien ne signale que le pipe avait pourtant
        # été ouvert avec succès.
        kernel32.CreateFileW.restype = wintypes.HANDLE
        kernel32.CreateFileW.argtypes = [
            wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD,
            wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE,
        ]
        kernel32.WaitNamedPipeW.restype = wintypes.BOOL
        kernel32.WaitNamedPipeW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD]
        kernel32.ReadFile.restype = wintypes.BOOL
        kernel32.ReadFile.argtypes = [
            wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD,
            ctypes.POINTER(wintypes.DWORD), wintypes.LPVOID,
        ]
        kernel32.WriteFile.restype = wintypes.BOOL
        kernel32.WriteFile.argtypes = [
            wintypes.HANDLE, wintypes.LPCVOID, wintypes.DWORD,
            ctypes.POINTER(wintypes.DWORD), wintypes.LPVOID,
        ]
        kernel32.CloseHandle.restype = wintypes.BOOL
        kernel32.CloseHandle.argtypes = [wintypes.HANDLE]

        # Un pipe dont toutes les instances sont déjà occupées fait échouer
        # CreateFile immédiatement (ERROR_PIPE_BUSY) : WaitNamedPipe patiente
        # qu'une instance se libère, dans la limite du timeout. Sans effet
        # (et sans erreur bloquante) si le pipe n'existe pas du tout —
        # CreateFile échouera juste ensuite avec ERROR_FILE_NOT_FOUND.
        kernel32.WaitNamedPipeW(_PIPE_NAME, timeout_ms)

        handle = kernel32.CreateFileW(
            _PIPE_NAME, GENERIC_READ | GENERIC_WRITE, 0, None,
            OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, None,
        )
        if not handle or handle == INVALID_HANDLE_VALUE:
            handle = None
            return None

        payload = (json.dumps(request) + "\n").encode("utf-8")
        written = wintypes.DWORD(0)
        ok = kernel32.WriteFile(handle, payload, len(payload),
                                ctypes.byref(written), None)
        if not ok:
            return None

        # Pipe en mode octet (pas mode message) des deux côtés : on lit
        # jusqu'à trouver la fin de ligne qui termine la réponse JSON.
        chunks: list[bytes] = []
        buf = ctypes.create_string_buffer(4096)
        read = wintypes.DWORD(0)
        while b"\n" not in b"".join(chunks):
            ok = kernel32.ReadFile(handle, buf, len(buf), ctypes.byref(read), None)
            if not ok or read.value == 0:
                break
            chunks.append(buf.raw[:read.value])

        data = b"".join(chunks)
        if not data:
            return None
        line = data.split(b"\n", 1)[0]
        resp = json.loads(line.decode("utf-8"))
        return resp if isinstance(resp, dict) else None
    except Exception:
        # Best-effort pur : n'importe quel pépin (pipe absent, ctypes,
        # décodage…) doit basculer l'appelant sur son chemin existant,
        # jamais lever jusqu'à lui.
        return None
    finally:
        if handle and kernel32:
            try:
                kernel32.CloseHandle(handle)
            except Exception:
                pass