"""Service Windows privilégié pour WGSecure. Installe/désinstalle le tunnel WireGuard, pose/retire les règles NRPT (split-DNS) et lit l'état du tunnel (`wg show`) sur demande de l'app — non privilégiée — pour éviter une invite d'authentification à chaque connexion et déconnexion. Tourne en LocalSystem. Exécutable séparé (`wgsecure-helper.exe`, build PyInstaller dédié) : jamais importé par l'app principale (`wgsecure.exe`), qui lui parle uniquement via le pipe nommé (`app/utils/helper_ipc.py`, ctypes pur, sans dépendance à pywin32). Invariant de sécurité central : le pipe reste accessible à tout process d'un utilisateur local, pas seulement WGSecure — donc CINQ commandes fixes seulement, jamais de script/commande arbitraire. Chaque commande valide strictement ses paramètres avant d'exécuter quoi que ce soit. """ from __future__ import annotations import ipaddress import json import logging import os import re import subprocess from logging.handlers import RotatingFileHandler import pywintypes import servicemanager import win32event import win32file import win32pipe import win32security import win32service import win32serviceutil from app.core.dns import _nrpt_add_split_dns_ps, _nrpt_remove_split_dns_ps from app.utils.platform_utils import get_wg_config_dir, wg_exe, wireguard_exe _PIPE_NAME = r"\\.\pipe\WGSecureHelper" # SYSTEM + Administrateurs : contrôle total. Utilisateurs interactifs # locaux (IU) : lecture/écriture seulement (se connecter, envoyer une # requête, lire la réponse) — jamais anonyme, jamais réseau (un pipe nommé # n'est de toute façon jamais accessible à distance sans partage explicite, # absent ici). _PIPE_SDDL = "D:(A;;GA;;;SY)(A;;GA;;;BA)(A;;GRGW;;;IU)" _INTERFACE_NAME_RE = re.compile(r"^[A-Za-z0-9_-]{1,15}$") _RULE_NAME_RE = re.compile(r"^WGSecure-SplitDNS-[A-Za-z0-9_-]+$") _HOSTNAME_LABEL = r"[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?" _HOSTNAME_RE = re.compile(rf"^{_HOSTNAME_LABEL}(\.{_HOSTNAME_LABEL})*$") _MAX_DOMAINS = 20 _LOG_DIR = os.path.join(os.environ.get("PROGRAMDATA", r"C:\ProgramData"), "WGSecure") _LOG_FILE = os.path.join(_LOG_DIR, "helper.log") def _get_logger() -> logging.Logger: os.makedirs(_LOG_DIR, exist_ok=True) logger = logging.getLogger("wgsecure-helper") if not logger.handlers: logger.setLevel(logging.INFO) handler = RotatingFileHandler(_LOG_FILE, maxBytes=1_000_000, backupCount=3, encoding="utf-8") handler.setFormatter(logging.Formatter( "%(asctime)s [%(levelname)s] %(message)s", "%Y-%m-%d %H:%M:%S")) logger.addHandler(handler) logger.propagate = False return logger class ValidationError(Exception): pass def _validate_interface_name(name) -> str: if not isinstance(name, str) or not _INTERFACE_NAME_RE.match(name): raise ValidationError(f"Nom d'interface invalide : {name!r}") return name def _validate_rule_name(name) -> str: if not isinstance(name, str) or not _RULE_NAME_RE.match(name): raise ValidationError(f"Nom de règle NRPT invalide : {name!r}") return name def _validate_domains(domains) -> list[str]: if not isinstance(domains, list) or not domains or len(domains) > _MAX_DOMAINS: raise ValidationError("Liste de domaines invalide") out = [] for d in domains: if not isinstance(d, str) or not _HOSTNAME_RE.match(d): raise ValidationError(f"Domaine invalide : {d!r}") out.append(d) return out def _validate_server(server) -> str: if not isinstance(server, str): raise ValidationError("Serveur invalide") try: ipaddress.ip_address(server) except ValueError: raise ValidationError(f"Adresse IP invalide : {server!r}") from None return server def _validate_config_path(config_path, name: str) -> str: """Le fichier doit résider exactement dans le dossier de configuration WireGuard et s'appeler .conf — jamais un chemin arbitraire. Le service ne lit ni n'écrit ce fichier ; il vérifie seulement qu'il pointe là où l'app (non privilégiée) est censée l'avoir déjà écrit avant de demander à wireguard.exe de l'installer. """ if not isinstance(config_path, str): raise ValidationError("Chemin de configuration invalide") expected = os.path.join(get_wg_config_dir(), f"{name}.conf") if os.path.normcase(os.path.abspath(config_path)) != os.path.normcase(os.path.abspath(expected)): raise ValidationError(f"Chemin de configuration hors du dossier attendu : {config_path!r}") return config_path def _run(cmd: list[str], timeout: int = 30) -> tuple[int, str]: """Exécution directe : le service tourne déjà en SYSTEM, aucune élévation à demander (contrairement à platform_utils.run_privileged, conçu pour un process non privilégié).""" try: result = subprocess.run(cmd, capture_output=True, timeout=timeout, encoding="utf-8", errors="replace") return result.returncode, (result.stderr.strip() or result.stdout.strip()) except subprocess.TimeoutExpired: return -1, "Timeout" except OSError as e: return -1, str(e) def _handle_install_tunnel(req: dict) -> dict: name = _validate_interface_name(req.get("name")) config_path = _validate_config_path(req.get("config_path"), name) code, msg = _run([wireguard_exe(), "/installtunnelservice", config_path], timeout=90) return {"ok": code == 0, "message": msg} def _handle_uninstall_tunnel(req: dict) -> dict: name = _validate_interface_name(req.get("name")) code, msg = _run([wireguard_exe(), "/uninstalltunnelservice", name], timeout=60) return {"ok": code == 0, "message": msg} def _handle_set_split_dns(req: dict) -> dict: server = _validate_server(req.get("server")) domains = _validate_domains(req.get("domains")) rule_name = _validate_rule_name(req.get("rule_name")) script = _nrpt_add_split_dns_ps(server, domains, rule_name) code, msg = _run(["powershell", "-NoProfile", "-NonInteractive", "-Command", script], timeout=20) return {"ok": code == 0, "message": msg} def _handle_clear_split_dns(req: dict) -> dict: rule_name = _validate_rule_name(req.get("rule_name")) script = _nrpt_remove_split_dns_ps(rule_name) code, msg = _run(["powershell", "-NoProfile", "-NonInteractive", "-Command", script], timeout=20) return {"ok": code == 0, "message": msg} def _handle_get_status(req: dict) -> dict: """Lecture seule : "message" porte la sortie de `wg show`, pas une erreur — "ok" reflète juste si la commande a pu s'exécuter. `latest_handshakes` : ajoute le sous-argument `latest-handshakes` (utilisé par `last_handshake_age` pour un horodatage exploitable, plutôt que le texte humain « il y a X secondes » de `wg show` simple). """ name = _validate_interface_name(req.get("name")) cmd = [wg_exe(), "show", name] if req.get("latest_handshakes"): cmd.append("latest-handshakes") code, msg = _run(cmd, timeout=10) return {"ok": code == 0, "message": msg} _HANDLERS = { "install_tunnel": _handle_install_tunnel, "uninstall_tunnel": _handle_uninstall_tunnel, "set_split_dns": _handle_set_split_dns, "clear_split_dns": _handle_clear_split_dns, "get_status": _handle_get_status, } def _create_pipe_instance(): sd = win32security.ConvertStringSecurityDescriptorToSecurityDescriptor( _PIPE_SDDL, win32security.SDDL_REVISION_1) sa = win32security.SECURITY_ATTRIBUTES() sa.SECURITY_DESCRIPTOR = sd return win32pipe.CreateNamedPipe( _PIPE_NAME, win32pipe.PIPE_ACCESS_DUPLEX, win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE | win32pipe.PIPE_WAIT, win32pipe.PIPE_UNLIMITED_INSTANCES, 65536, 65536, 0, sa, ) def _handle_connection(handle, logger: logging.Logger) -> None: resp: dict try: # Un seul message par connexion (le client ferme après avoir lu la # réponse) : bornée à 64 Ko pour ne jamais bloquer sur un client # qui n'enverrait jamais de fin de ligne. data = b"" for _ in range(16): _, chunk = win32file.ReadFile(handle, 4096) data += chunk if b"\n" in data or not chunk: break line = data.split(b"\n", 1)[0] req = json.loads(line.decode("utf-8")) cmd = req.get("cmd") if isinstance(req, dict) else None handler = _HANDLERS.get(cmd) if handler is None: resp = {"ok": False, "message": f"Commande inconnue : {cmd!r}"} else: try: resp = handler(req) except ValidationError as e: resp = {"ok": False, "message": str(e)} logger.info("%s -> ok=%s", cmd, resp.get("ok")) except Exception as e: logger.exception("Requête invalide ou erreur pipe") resp = {"ok": False, "message": f"Erreur interne : {e}"} try: win32file.WriteFile(handle, (json.dumps(resp) + "\n").encode("utf-8")) except Exception: pass class WGSecureHelperService(win32serviceutil.ServiceFramework): _svc_name_ = "WGSecureHelper" _svc_display_name_ = "WGSecure Helper" _svc_description_ = ( "Applique pour WGSecure, sans invite d'authentification répétée, " "l'installation du tunnel WireGuard et les règles DNS (split-DNS) " "qui exigent normalement des droits administrateur." ) def __init__(self, args): win32serviceutil.ServiceFramework.__init__(self, args) self._stop_event = win32event.CreateEvent(None, 0, 0, None) self._running = True def SvcStop(self): self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING) self._running = False # Débloque un ConnectNamedPipe() en attente en se connectant à son # propre pipe — plus simple et plus robuste qu'un I/O recouvrant # pour un service à trafic très faible (quelques appels par # connexion/déconnexion de tunnel). try: h = win32file.CreateFile( _PIPE_NAME, win32file.GENERIC_READ, 0, None, win32file.OPEN_EXISTING, 0, None) win32file.CloseHandle(h) except pywintypes.error: pass win32event.SetEvent(self._stop_event) def SvcDoRun(self): servicemanager.LogMsg( servicemanager.EVENTLOG_INFORMATION_TYPE, servicemanager.PYS_SERVICE_STARTED, (self._svc_name_, ""), ) logger = _get_logger() logger.info("Service démarré") try: self._serve_forever(logger) except Exception: # Filet de sécurité final : si malgré tout ce qui suit une # exception s'échappe encore, elle laisse au moins une trace # exploitable ici plutôt qu'un service qui disparaît en # silence sans rien dans le log. logger.exception("Arrêt inattendu de la boucle principale") finally: logger.info("Service arrêté") def _serve_forever(self, logger: logging.Logger) -> None: # Chaque étape est individuellement protégée et ne peut donc jamais # faire sortir la boucle sur une exception non prévue : un service # censé tourner indéfiniment sans surveillance ne doit jamais # s'arrêter à cause d'un incident sur UNE requête. Une version # antérieure ne capturait que pywintypes.error (pas Exception en # général) et ne protégeait pas CloseHandle individuellement dans # le `finally` : une erreur non prévue là pouvait remonter et # arrêter tout le service — expliquant un premier échange réussi # (juste après le démarrage du service) puis silence total ensuite. ERROR_PIPE_CONNECTED = 535 while self._running: handle = None try: handle = _create_pipe_instance() try: win32pipe.ConnectNamedPipe(handle, None) except pywintypes.error as e: # Le client s'est connecté entre CreateNamedPipe et # ConnectNamedPipe : cas normal (pas une erreur), pas # rare vu la fréquence des appels — sans ce cas # particulier, la requête était silencieusement ratée # à chaque fois qu'il se produisait. if e.winerror != ERROR_PIPE_CONNECTED: raise if not self._running: break _handle_connection(handle, logger) except Exception: logger.exception("Erreur sur une connexion cliente") finally: if handle: try: win32pipe.DisconnectNamedPipe(handle) except Exception: pass try: win32file.CloseHandle(handle) except Exception: pass if __name__ == "__main__": win32serviceutil.HandleCommandLine(WGSecureHelperService)