Enclume/enclume/auth.py

"""Connexion optionnelle par OpenID Connect.

Le site reste entierement utilisable sans compte : se connecter n'ajoute que
l'enregistrement de projets cote serveur et la soumission de packs.

Un seul fournisseur est declare, Authentik, qui federe lui-meme GitHub, Google
ou tout autre fournisseur social. L'application n'a donc jamais a connaitre les
identifiants de ces services, ni a etre redeployee quand on en ajoute un.
"""

from __future__ import annotations

import functools
from typing import Any, Callable

from authlib.integrations.flask_client import OAuth
from flask import Blueprint, current_app, jsonify, redirect, request, session, url_for

from .config import Config
from .db import Utilisateur, db, maintenant

oauth = OAuth()
bp = Blueprint("auth", __name__)


def init_app(app) -> None:
    oauth.init_app(app)
    if not Config.oidc_actif():
        return
    oauth.register(
        name="fournisseur",
        client_id=Config.OIDC_CLIENT_ID,
        client_secret=Config.OIDC_CLIENT_SECRET,
        server_metadata_url=Config.OIDC_METADATA,
        client_kwargs={"scope": Config.OIDC_SCOPES},
    )
    app.register_blueprint(bp)


# --------------------------------------------------------------------------- #
# Etat de la session
# --------------------------------------------------------------------------- #

def utilisateur_courant() -> Utilisateur | None:
    identifiant = session.get("utilisateur_id")
    if not identifiant or not Config.base_active():
        return None
    return db.session.get(Utilisateur, identifiant)


def est_administrateur(utilisateur: Utilisateur | None = None) -> bool:
    utilisateur = utilisateur or utilisateur_courant()
    return bool(utilisateur and utilisateur.administrateur)


def connexion_requise(vue: Callable) -> Callable:
    @functools.wraps(vue)
    def enveloppe(*args: Any, **kwargs: Any):
        if utilisateur_courant() is None:
            return jsonify({"erreur": "connexion requise"}), 401
        return vue(*args, **kwargs)

    return enveloppe


def administration_requise(vue: Callable) -> Callable:
    @functools.wraps(vue)
    def enveloppe(*args: Any, **kwargs: Any):
        utilisateur = utilisateur_courant()
        if utilisateur is None:
            return jsonify({"erreur": "connexion requise"}), 401
        if not utilisateur.administrateur:
            return jsonify({"erreur": "reserve a la moderation"}), 403
        return vue(*args, **kwargs)

    return enveloppe


# --------------------------------------------------------------------------- #
# Parcours de connexion
# --------------------------------------------------------------------------- #

def _destination_sure(brut: str | None) -> str:
    """N'accepte qu'un chemin interne : pas de redirection ouverte."""
    if not brut or not brut.startswith("/") or brut.startswith("//"):
        return "/"
    return brut


@bp.get("/connexion")
def connexion():
    session["retour"] = _destination_sure(request.args.get("retour"))
    redirection = url_for("auth.retour", _external=True)
    if Config.URL_PUBLIQUE:
        redirection = f"{Config.URL_PUBLIQUE}{url_for('auth.retour')}"
    return oauth.fournisseur.authorize_redirect(redirection)


@bp.get("/connexion/retour")
def retour():
    try:
        jeton = oauth.fournisseur.authorize_access_token()
    except Exception:  # noqa: BLE001 - echec d'echange, on renvoie sans detail
        current_app.logger.warning("echec de l'echange du jeton OIDC")
        return redirect("/?connexion=echec")

    revendications = jeton.get("userinfo") or {}
    if not revendications:
        try:
            revendications = oauth.fournisseur.userinfo(token=jeton)
        except Exception:  # noqa: BLE001
            revendications = {}

    sujet = str(revendications.get("sub") or "").strip()
    if not sujet:
        return redirect("/?connexion=echec")

    utilisateur = db.session.query(Utilisateur).filter_by(sujet=sujet).one_or_none()
    if utilisateur is None:
        utilisateur = Utilisateur(sujet=sujet)
        db.session.add(utilisateur)

    utilisateur.email = str(revendications.get("email") or "")[:320]
    utilisateur.nom = str(
        revendications.get("name") or revendications.get("preferred_username") or ""
    )[:255]
    utilisateur.administrateur = _est_admin(revendications)
    utilisateur.vu_le = maintenant()
    db.session.commit()

    session.clear()
    session["utilisateur_id"] = utilisateur.id
    session.permanent = True
    return redirect(_destination_sure(session.pop("retour", "/")))


def _est_admin(revendications: dict) -> bool:
    """Deux voies : appartenance a un groupe, ou liste explicite de comptes.

    La seconde existe pour le premier demarrage, avant que le mapping de portee
    "groups" ne soit configure cote Authentik.
    """
    groupes = revendications.get("groups") or []
    if isinstance(groupes, str):
        groupes = [groupes]
    if Config.ADMIN_GROUPE and Config.ADMIN_GROUPE in groupes:
        return True
    email = str(revendications.get("email") or "").lower()
    return bool(email and email in Config.ADMIN_COMPTES)


@bp.post("/deconnexion")
def deconnexion():
    session.clear()
    return jsonify({"statut": "deconnecte"})