WGSecure/app/core/vpn_session.py

"""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}")