tuxgyverandClaude Opus 5
a056ad260a
prepare_quit() attend les threads en cours avant toute destruction, mais sa liste en oubliait deux : le sondage d'état ajouté en 0.8.6 — ma régression — et le thread de validation de la fenêtre de saisie, parenté à cette fenêtre depuis la v0.8.2 et jamais attendu. Fermer la fenêtre de code pendant qu'une requête est en vol suffisait à avorter le process, ce qui arrive dès que le serveur tarde. Reproduit avec un serveur qui accepte sans jamais répondre, puis vérifié corrigé sur les deux chemins. Le sondage est ramené à 4 s : l'arrêt l'attend désormais, et dix secondes y seraient une interface figée. Son échec ne coûte rien, contrairement à une validation ratée qui gâche un code déjà saisi. Ajoute le temps restant sur la ligne Accès distant, et un repli pour les serveurs sans GET /api/session : sur un 404, le client observe si le trafic atteint le réseau distant plutôt que de réclamer un code déjà validé. Un 403 reste une réponse qui fait autorité. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
231 lines
10 KiB
Python
231 lines
10 KiB
Python
"""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
|
|
|
|
_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
|
|
|
|
|
|
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.
|
|
"""
|
|
|
|
ok: bool
|
|
message: str
|
|
access_until: datetime | None = None
|
|
|
|
|
|
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 de ce réseau (10.6.0.1 pour un client
|
|
en 10.6.0.6). 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.
|
|
"""
|
|
explicit = (cfg.wg.get("api_url") or "").strip()
|
|
if explicit:
|
|
return explicit.rstrip("/")
|
|
|
|
client_address = (cfg.wg.get("client_address") or "").split("/")[0].strip()
|
|
try:
|
|
host = ipaddress.ip_address(client_address)
|
|
except ValueError:
|
|
return ""
|
|
gateway = ipaddress.ip_address(int(host) - (int(host) & 0xFF) + 1)
|
|
return f"http://{gateway}:8000"
|
|
|
|
|
|
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é ».
|
|
|
|
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:
|
|
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 ne distingue pas « accès fermé » de « serveur injoignable » :
|
|
l'appelant traite les deux pareil — il redemande un code, ce qui est le
|
|
comportement sûr dans le doute.
|
|
"""
|
|
base = api_base_url(cfg)
|
|
if not base:
|
|
return AuthResult(False, "Adresse du serveur d'authentification inconnue.")
|
|
|
|
request = urllib.request.Request(f"{base}/api/session", method="GET")
|
|
try:
|
|
with urllib.request.urlopen(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)")
|
|
return AuthResult(False, detail or f"Accès fermé (code {exc.code})")
|
|
except (urllib.error.URLError, OSError, ValueError) as exc:
|
|
# Serveur ancien (route absente), injoignable, ou réponse illisible :
|
|
# aucune information, donc aucune conclusion — on ne prétend pas que
|
|
# l'accès est ouvert.
|
|
return AuthResult(False, f"État de l'accès inconnu : {exc}")
|
|
|
|
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 urllib.request.urlopen(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}")
|