tuxgyverandClaude Sonnet 5
7beb6deb7a
Ajoute wgsecure-helper, un service Windows privilégié (LocalSystem) qui installe/désinstalle le tunnel et pose/retire les règles NRPT sans invite d'authentification répétée. L'ACL du service WireGuard (v0.7.6) ne délègue que démarrer/arrêter un tunnel déjà installé, jamais le créer/supprimer, ce que WGSecure fait pourtant à chaque connexion/déconnexion — d'où les prompts qui persistaient malgré v0.7.6-v0.7.11. Corrige aussi le handshake, qui échouait pour la même raison (wg show sans élévation complète). Purement additif : l'app retombe sur son chemin élevé existant si le service est absent (composant optionnel dans l'installeur). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
"""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
|
|
|
|
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
|
|
kernel32.CreateFileW.restype = wintypes.HANDLE
|
|
kernel32.CreateFileW.argtypes = [
|
|
wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD,
|
|
wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, 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.
|
|
try:
|
|
kernel32.WaitNamedPipeW(_PIPE_NAME, timeout_ms)
|
|
except OSError:
|
|
pass
|
|
|
|
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:
|
|
return None
|
|
|
|
try:
|
|
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 (OSError, ValueError):
|
|
return None
|
|
finally:
|
|
kernel32.CloseHandle(handle)
|