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 urllib.error
import urllib.request
from datetime import datetime, timezone
from typing import NamedTuple

from app.core.config import Config

_TIMEOUT = 10


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