"""Authentification auprès du serveur, après montage du tunnel. Le tunnel qui monte ne donne accès à rien : le serveur place le pair en quarantaine et ne laisse passer son trafic vers le réseau distant qu'une fois un code à 6 chiffres validé. C'est ce que fait ce module. Le code est vérifié **par le serveur**, contre un secret que cette application ne détient pas — contrairement au MFA local d'origine, qui validait un code qu'elle pouvait elle-même produire, et ne verrouillait donc que sa propre interface. L'ordre est imposé par le réseau : l'API du serveur n'est joignable que depuis l'intérieur du tunnel. On monte donc d'abord, on s'authentifie ensuite — le tunnel jouant le rôle d'un réseau de quarantaine tant que le code n'est pas passé. `urllib` plutôt qu'une bibliothèque HTTP tierce : un seul appel POST, aucune raison d'alourdir les dépendances de l'application. """ from __future__ import annotations import ipaddress import json import socket import urllib.error import urllib.request from datetime import datetime, timezone from typing import NamedTuple from app.core.config import Config from app.core.wireguard import tunnel_gateway _TIMEOUT = 10 # Le sondage d'état est borné plus court que la validation d'un code : il part # tout seul au montage du tunnel, et l'arrêt de l'application l'attend avant de # se terminer — dix secondes d'attente y seraient une interface figée. Son # échec ne coûte rien (le bouton reste disponible), là où une validation ratée # gâche un code déjà saisi. _PROBE_TIMEOUT = 4 # Sonde d'accessibilité du réseau distant : un aller-retour sur le LAN passé # le tunnel, deux secondes suffisent largement — au-delà, c'est filtré. _REACH_TIMEOUT = 2 # Ouvreur sans proxy. `urllib.request.urlopen` emploie l'ouvreur par défaut, # dont le `ProxyHandler` lit `http_proxy` / `HTTP_PROXY` dans l'environnement : # sur un poste d'entreprise, la requête vers l'adresse privée du tunnel partait # au proxy au lieu du serveur. Sa réponse — souvent un 404 — était prise pour # celle du serveur, et un 404 déclenche justement le repli d'observation # réseau. Symptôme parfait de panne intermittente : selon l'environnement de # lancement, ça passait ou non. Le serveur d'authentification n'est de toute # façon joignable que par l'intérieur du tunnel, jamais par un proxy. _OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({})) class AuthResult(NamedTuple): """Réponse du serveur à une demande d'ouverture d'accès. `access_until` porte l'échéance telle que le serveur l'annonce — il en est la seule autorité, le client la relit sans jamais la calculer. `None` quand aucune échéance n'est annoncée : l'accès est alors ouvert sans terme connu, et le client s'abstient d'en inventer un. `authoritative` dit si le serveur a réellement répondu. Un `ok` faux couvrait deux situations très différentes — « le serveur refuse » et « le serveur n'a rien dit » — que l'appelant traitait pareil : un hoquet réseau au montage du tunnel faisait donc surgir la fenêtre de saisie et brûler un code, alors que l'accès était peut-être déjà ouvert. Faux ici veut dire « on ne sait pas » : serveur injoignable, réponse illisible, ou route `/api/session` absente d'un serveur antérieur à la v0.3.0. """ ok: bool message: str access_until: datetime | None = None authoritative: bool = True def _parse_until(raw: object) -> datetime | None: """Lit l'échéance renvoyée par le serveur, en UTC. Une valeur absente ou illisible ne vaut pas refus : l'accès vient d'être ouvert. Elle rend seulement l'échéance inconnue, cas déjà prévu. """ if not isinstance(raw, str) or not raw: return None try: moment = datetime.fromisoformat(raw.replace("Z", "+00:00")) except ValueError: return None # Le serveur date en UTC ; sans fuseau explicite on l'y suppose, plutôt # que de prendre l'horodatage pour une heure locale et de décaler # l'échéance de l'offset du poste. return moment if moment.tzinfo else moment.replace(tzinfo=timezone.utc) def api_base_url(cfg: Config) -> str: """URL de l'API du serveur, telle qu'atteignable depuis le tunnel. À défaut d'URL explicite, elle est déduite de l'adresse du client : le serveur occupe la première adresse du réseau du tunnel. C'est l'adresse la plus fiable depuis la quarantaine — elle désigne le serveur lui-même, dont l'accès n'est jamais filtré, alors qu'une adresse du réseau distant le serait. La déduction est déléguée à `wireguard.tunnel_gateway`, qui sert déjà au diagnostic. Elle se faisait ici à part, en retranchant le dernier octet : cela supposait un /24 et ignorait les IPs autorisées, alors qu'une adresse client en /32 — la forme la plus courante — ne décrit aucun réseau. Un client en 10.6.1.7/32 sur un réseau 10.6.0.0/16 se voyait ainsi expédié vers 10.6.1.1, qui n'existe pas : sondage et validation échouaient tous deux, sans autre issue que de renseigner `api_url` à la main. """ explicit = (cfg.wg.get("api_url") or "").strip() if explicit: # Le schéma est vérifié plutôt que repris tel quel : identifiant et # code partent dans cette URL, et une valeur saisie de travers — un # hôte nu sans schéma, un `file:` — enverrait la requête ailleurs que # prévu, ou la ferait échouer sans dire pourquoi. `https` est accepté # pour un serveur qui l'expose ; `http` reste le cas courant, la # confidentialité venant alors du tunnel lui-même. if not explicit.lower().startswith(("http://", "https://")): return "" return explicit.rstrip("/") gateway = tunnel_gateway(cfg.wg.get("client_address") or "", cfg.wg.get("allowed_ips") or "") if not gateway: return "" # Une adresse IPv6 doit être encadrée de crochets dans une URL, sans quoi # son propre « : » se confond avec celui du port. host = f"[{gateway}]" if ":" in gateway else gateway return f"http://{host}:8000" def _routed_through_tunnel(cfg: Config, target: str) -> bool: """Le trafic vers `target` sort-il bien par le tunnel ? Rien ne garantit que le serveur DNS déclaré ne soit joignable *que* par le tunnel : une adresse RFC1918 d'entreprise peut très bien être routée par le réseau local du poste — télétravail depuis un site jumeau, VPN d'un autre fournisseur déjà monté. La sonde d'accessibilité réussissait alors hors tunnel et déclarait la quarantaine levée alors qu'elle tenait toujours : l'accès s'affichait « Ouvert ✓ », sans échéance donc sans fin, et le bouton de saisie restait masqué. Le repli censé éviter le blocage le provoquait. La question se tranche sans émettre le moindre paquet : `connect()` sur une socket UDP ne fait que fixer la destination, et le noyau y attache déjà l'adresse source qu'il emploierait — celle du tunnel si et seulement si c'est par là que passe la route. """ client_ip = ((cfg.wg.get("client_address") or "") .split(",")[0].split("/")[0].strip()) if not client_ip: return False try: family, socktype, proto, _, sockaddr = socket.getaddrinfo( target, 53, type=socket.SOCK_DGRAM)[0] probe = socket.socket(family, socktype, proto) try: probe.connect(sockaddr) source = probe.getsockname()[0] finally: probe.close() except (OSError, IndexError): return False # Comparaison numérique : le noyau peut rendre une forme différente de # celle écrite en configuration (zéros abrégés d'une adresse IPv6). try: return ipaddress.ip_address(source) == ipaddress.ip_address(client_ip) except ValueError: return False def _remote_network_reachable(cfg: Config) -> bool: """Le trafic passe-t-il réellement vers le réseau distant ? Repli pour les serveurs antérieurs à la v0.3.0, qui n'exposent pas `GET /api/session` : sans cette route le client ne peut pas *demander* son état, il ne lui reste qu'à l'observer. La cible est le serveur DNS du split-DNS, qui vit sur le réseau distant : la quarantaine jette le trafic *traversant* (chaîne FORWARD), donc une réponse quelconque de sa part prouve que la quarantaine est levée. Un refus de connexion en est une : le paquet a atteint l'hôte, qui a répondu par un RST. Seul un silence signifie « filtré ». Encore faut-il que la cible soit atteinte *par le tunnel* — sans quoi on observe le réseau local et on en conclut n'importe quoi (cf. `_routed_through_tunnel`). Renvoie False dès qu'aucune cible n'est configurée : sans point de comparaison, on n'observe rien et on ne conclut rien. """ target = (cfg.wg.get("split_dns_server") or "").strip() if not target or not _routed_through_tunnel(cfg, target): return False try: socket.create_connection((target, 53), timeout=_REACH_TIMEOUT).close() return True except ConnectionRefusedError: return True # l'hôte a répondu : le trafic passe except OSError: return False # silence, ou réseau injoignable def session_status(cfg: Config) -> AuthResult: """Demande au serveur si l'accès de ce poste est déjà ouvert. Un tunnel qui tombe et remonte ne révoque rien côté serveur : l'autorisation porte sur l'adresse du pair, et cette adresse ne change pas. Le client, lui, oubliait tout à chaque reconnexion et réclamait un code déjà validé — c'est ce que cette question évite. Aucun identifiant n'est envoyé : le serveur reconnaît l'appelant à l'IP source du tunnel, et ne renseigne donc jamais sur un autre compte que celui qui parle. Rien à demander avant que le tunnel ne soit monté, l'API n'étant joignable que par l'intérieur. Un `ok` faux vaut « accès non ouvert » ; c'est `authoritative` qui dit si le serveur l'a affirmé ou si l'on n'en sait rien. L'affichage reste prudent dans les deux cas — le bouton de saisie demeure atteignable — mais l'appelant ne fait plus surgir la fenêtre sur un simple silence du réseau. """ base = api_base_url(cfg) if not base: return AuthResult(False, "Adresse du serveur d'authentification " "inconnue.", None, False) request = urllib.request.Request(f"{base}/api/session", method="GET") try: with _OPENER.open(request, timeout=_PROBE_TIMEOUT) as response: body = json.loads(response.read().decode("utf-8")) until = _parse_until(body.get("access_until")) if until is None: return AuthResult(True, "Accès réseau déjà ouvert") return AuthResult(True, "Accès réseau déjà ouvert jusqu'à " f"{until.astimezone():%d/%m/%Y %H:%M}", until) except urllib.error.HTTPError as exc: try: detail = json.loads(exc.read().decode("utf-8")).get("detail", "") except (ValueError, OSError): detail = "" if exc.code == 404 and _remote_network_reachable(cfg): # Serveur antérieur à la v0.3.0 : la route n'existe pas, mais le # trafic atteint le réseau distant — l'accès est donc bien ouvert. # Aucune échéance en revanche : seul le serveur la connaît, et il # n'a pas de quoi la dire. Mieux vaut un décompte absent qu'un # bouton qui réclame un code déjà validé. return AuthResult(True, "Accès réseau ouvert (constaté sur le " "réseau ; serveur trop ancien pour " "l'annoncer)") if exc.code == 404: # Route absente et rien d'observable sur le réseau : c'est un # serveur trop ancien pour répondre à la question, pas un refus. return AuthResult(False, detail or "État de l'accès inconnu " "(serveur sans /api/session)", None, False) # Toute autre réponse HTTP vient du serveur lui-même : elle fait # autorité, un 403 comme le reste. return AuthResult(False, detail or f"Accès fermé (code {exc.code})") except (urllib.error.URLError, OSError, ValueError) as exc: # Serveur injoignable ou réponse illisible : aucune information, donc # aucune conclusion — on ne prétend pas que l'accès est ouvert, mais # on ne prétend pas non plus qu'il est fermé. return AuthResult(False, f"État de l'accès inconnu : {exc}", None, False) def authenticate(cfg: Config, code: str, username: str = "") -> AuthResult: """Valide un code TOTP auprès du serveur et ouvre l'accès réseau. Retourne un `AuthResult`. Un échec n'est jamais silencieux : sans cette validation le tunnel reste monté mais ne mène nulle part, et l'utilisateur doit comprendre pourquoi plutôt que de constater un réseau inexplicablement muet. `username` permet de fournir l'identifiant à la volée, quand il n'est pas encore en configuration : la fenêtre de saisie le demande alors avec le code, plutôt que de renvoyer l'utilisateur vers le panneau Administrateur au moment précis où il cherche à ouvrir son accès. Il est enregistré une fois le serveur l'ayant accepté. """ base = api_base_url(cfg) if not base: return AuthResult(False, "Adresse du serveur d'authentification " "inconnue. Renseignez-la dans le panneau " "Administrateur.") username = (username or cfg.wg.get("vpn_username") or "").strip() if not username: return AuthResult(False, "Indiquez votre identifiant VPN.") payload = json.dumps({"username": username, "code": code}).encode("utf-8") request = urllib.request.Request( f"{base}/api/session", data=payload, headers={"Content-Type": "application/json"}, method="POST", ) try: with _OPENER.open(request, timeout=_TIMEOUT) as response: body = json.loads(response.read().decode("utf-8")) # Affichée dans le fuseau du poste : l'échéance arrive en UTC, et la # présenter telle quelle annonçait une heure fausse à l'utilisateur. until = _parse_until(body.get("access_until")) if until is None: return AuthResult(True, "Accès réseau ouvert") return AuthResult(True, "Accès réseau ouvert jusqu'à " f"{until.astimezone():%d/%m/%Y %H:%M}", until) except urllib.error.HTTPError as exc: # Le serveur explique lui-même le refus (code invalide, compte # verrouillé, expiré…) : le relayer tel quel vaut mieux qu'un # message générique qui obligerait à consulter les journaux. try: detail = json.loads(exc.read().decode("utf-8")).get("detail", "") except (ValueError, OSError): detail = "" return AuthResult(False, detail or f"Authentification refusée (code {exc.code})") except urllib.error.URLError as exc: return AuthResult(False, f"Serveur d'authentification injoignable " f"({base}) : {exc.reason}") except (OSError, ValueError) as exc: return AuthResult(False, f"Échec de l'authentification : {exc}")