"""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 (pywintypes.error, ValueError, OSError) 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 pywintypes.error: 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é") while self._running: try: handle = _create_pipe_instance() except pywintypes.error: logger.exception("Création du pipe impossible") break try: win32pipe.ConnectNamedPipe(handle, None) if not self._running: break _handle_connection(handle, logger) except pywintypes.error: logger.exception("Erreur sur une connexion cliente") finally: try: win32pipe.DisconnectNamedPipe(handle) except pywintypes.error: pass win32file.CloseHandle(handle) logger.info("Service arrêté") if __name__ == "__main__": win32serviceutil.HandleCommandLine(WGSecureHelperService)