344 lines
13 KiB
Python
344 lines
13 KiB
Python
import base64
|
|
import io
|
|
import os
|
|
import bcrypt
|
|
import pyotp
|
|
import qrcode
|
|
import yaml
|
|
import reflex as rx
|
|
from pathlib import Path
|
|
|
|
DATA_DIR = Path(os.getenv("DATA_DIR", "data"))
|
|
TOTP_ISSUER = "EPTM Dashboard"
|
|
|
|
|
|
def _load_auth_full() -> dict:
|
|
"""Lit auth.yaml complet (config dict, pas seulement usernames)."""
|
|
auth_file = DATA_DIR / "auth.yaml"
|
|
if auth_file.exists():
|
|
with open(auth_file, encoding="utf-8") as f:
|
|
return yaml.safe_load(f) or {}
|
|
return {}
|
|
|
|
|
|
def _save_auth_full(cfg: dict) -> None:
|
|
auth_file = DATA_DIR / "auth.yaml"
|
|
with open(auth_file, "w", encoding="utf-8") as f:
|
|
yaml.dump(cfg, f, allow_unicode=True)
|
|
|
|
|
|
def _make_totp_qr_data_url(secret: str, label: str) -> str:
|
|
"""Génère un data URL PNG du QR code TOTP."""
|
|
uri = pyotp.totp.TOTP(secret).provisioning_uri(name=label, issuer_name=TOTP_ISSUER)
|
|
img = qrcode.make(uri)
|
|
buf = io.BytesIO()
|
|
img.save(buf, format="PNG")
|
|
return f"data:image/png;base64,{base64.b64encode(buf.getvalue()).decode()}"
|
|
|
|
|
|
class AuthState(rx.State):
|
|
# Persisted in browser localStorage (survives hot reload / container restart).
|
|
# Note: client-side trustable only because re-validated against auth.yaml in check_auth.
|
|
username: str = rx.LocalStorage("", sync=True)
|
|
name: str = rx.LocalStorage("", sync=True)
|
|
role: str = rx.LocalStorage("user", sync=True)
|
|
photo_url: str = rx.LocalStorage("", sync=True)
|
|
# Thème de couleur de l'interface : "eptm" (défaut), "bleu", "indigo", "vert".
|
|
# Appliqué via data-theme sur <html> côté client.
|
|
theme: str = rx.LocalStorage("eptm", sync=True)
|
|
|
|
# In-memory only (login form, transient UI state)
|
|
login_user: str = ""
|
|
login_pass: str = ""
|
|
login_error: str = ""
|
|
|
|
# 2FA flow (in-memory, ephémère)
|
|
# totp_step in {"password", "setup", "verify"}
|
|
totp_step: str = "password"
|
|
totp_pending_user: str = ""
|
|
totp_secret_pending: str = "" # secret généré en setup, pas encore sauvé
|
|
totp_qr_data_url: str = ""
|
|
totp_code: str = ""
|
|
totp_error: str = ""
|
|
|
|
sidebar_collapsed: bool = False
|
|
mobile_menu_open: bool = False
|
|
admin_expanded: bool = True
|
|
doc_expanded: bool = False
|
|
# Compteur de messages feedback "new" (admin uniquement)
|
|
feedback_new_count: int = 0
|
|
# Flag : True si le user (non-admin) n'a aucune classe accordée et doit
|
|
# passer par l'enrôlement Escada. Auto-set par check_auth.
|
|
must_enroll: bool = False
|
|
# L'user a fermé le popup manuellement (pour cette session) — on ne le
|
|
# réaffiche pas automatiquement même si must_enroll reste True.
|
|
enroll_dismissed: bool = False
|
|
# Données de l'utilisateur connecté pour la section enrôlement Escada
|
|
# (rechargées à chaque check_auth pour éviter la pollution entre sessions).
|
|
my_classes: list[str] = []
|
|
classes_unknown: list[str] = []
|
|
escada_username: str = ""
|
|
escada_has_password: bool = False
|
|
|
|
@rx.var
|
|
def authenticated(self) -> bool:
|
|
return self.username != ""
|
|
|
|
@rx.var
|
|
def name_initials(self) -> str:
|
|
if not self.name:
|
|
return "?"
|
|
parts = self.name.split()
|
|
if len(parts) >= 2:
|
|
return (parts[0][0] + parts[1][0]).upper()
|
|
return self.name[:2].upper()
|
|
|
|
def toggle_sidebar(self):
|
|
self.sidebar_collapsed = not self.sidebar_collapsed
|
|
|
|
def toggle_mobile_menu(self):
|
|
self.mobile_menu_open = not self.mobile_menu_open
|
|
|
|
def close_mobile_menu(self):
|
|
self.mobile_menu_open = False
|
|
|
|
def toggle_admin(self):
|
|
self.admin_expanded = not self.admin_expanded
|
|
|
|
def toggle_doc(self):
|
|
self.doc_expanded = not self.doc_expanded
|
|
|
|
def set_login_user(self, value: str):
|
|
self.login_user = value
|
|
|
|
def set_login_pass(self, value: str):
|
|
self.login_pass = value
|
|
|
|
def set_totp_code(self, value: str):
|
|
# Garder uniquement les chiffres, max 6
|
|
self.totp_code = "".join(ch for ch in value if ch.isdigit())[:6]
|
|
self.totp_error = ""
|
|
|
|
def index_redirect(self):
|
|
if self.authenticated:
|
|
return rx.redirect("/accueil")
|
|
return rx.redirect("/login")
|
|
|
|
def redirect_if_authenticated(self):
|
|
if self.authenticated:
|
|
return rx.redirect("/accueil")
|
|
|
|
def check_auth(self):
|
|
if not self.username:
|
|
return rx.redirect("/login")
|
|
users = self._load_users()
|
|
if self.username not in users:
|
|
self._clear_session()
|
|
return rx.redirect("/login")
|
|
self.photo_url = users[self.username].get("avatar_url", "")
|
|
# Re-synchronise le thème depuis auth.yaml (au cas où changé sur un autre device).
|
|
stored_theme = users[self.username].get("theme") or "eptm"
|
|
if stored_theme != self.theme:
|
|
self.theme = stored_theme
|
|
# Compteur feedback (admin uniquement) — pour le badge sidebar
|
|
self._refresh_feedback_count()
|
|
# Recharge les données d'enrôlement depuis auth.yaml à chaque page —
|
|
# évite la pollution si un autre user était dans la session avant.
|
|
u = users.get(self.username) or {}
|
|
self.my_classes = list(u.get("allowed_classes") or [])
|
|
self.escada_username = u.get("escada_username") or ""
|
|
self.escada_has_password = bool(u.get("escada_password"))
|
|
self.classes_unknown = [] # reset l'avertissement à chaque page
|
|
# Détecte si l'user doit s'enrôler (non-admin sans classes accordées)
|
|
if u.get("role") == "admin":
|
|
self.must_enroll = False
|
|
else:
|
|
self.must_enroll = not self.my_classes
|
|
return self._apply_theme_script(self.theme)
|
|
|
|
def dismiss_enroll(self):
|
|
"""Ferme le popup d'enrôlement pour la session courante."""
|
|
self.enroll_dismissed = True
|
|
|
|
def _refresh_feedback_count(self):
|
|
if self.role != "admin":
|
|
self.feedback_new_count = 0
|
|
return
|
|
try:
|
|
from src.db import get_session, FeedbackMessage
|
|
from sqlalchemy import select, func
|
|
sess = get_session()
|
|
try:
|
|
self.feedback_new_count = sess.execute(
|
|
select(func.count(FeedbackMessage.id))
|
|
.where(FeedbackMessage.status == "new")
|
|
).scalar() or 0
|
|
finally:
|
|
sess.close()
|
|
except Exception:
|
|
self.feedback_new_count = 0
|
|
|
|
@staticmethod
|
|
def _apply_theme_script(theme: str):
|
|
"""Script JS qui set data-theme + color-scheme sur <html> immédiatement
|
|
(sans attendre re-render). color-scheme empêche le browser de bascule
|
|
dark sur OS dark mode."""
|
|
safe = "".join(c for c in (theme or "eptm") if c.isalnum() or c in "-_")
|
|
scheme = "dark" if safe == "sombre" else "light"
|
|
if not safe or safe == "eptm":
|
|
return rx.call_script(
|
|
"document.documentElement.removeAttribute('data-theme');"
|
|
"document.body && document.body.removeAttribute('data-theme');"
|
|
f"document.documentElement.style.colorScheme = '{scheme}';"
|
|
)
|
|
return rx.call_script(
|
|
f"document.documentElement.setAttribute('data-theme', '{safe}');"
|
|
f"document.body && document.body.setAttribute('data-theme', '{safe}');"
|
|
f"document.documentElement.style.colorScheme = '{scheme}';"
|
|
)
|
|
|
|
def set_theme(self, value: str):
|
|
"""Change le thème de couleur (persiste en LocalStorage et auth.yaml)."""
|
|
if value not in ("eptm", "bleu", "indigo", "vert", "sombre"):
|
|
value = "eptm"
|
|
self.theme = value
|
|
# Persister dans auth.yaml pour synchronisation multi-device.
|
|
if self.username:
|
|
cfg = _load_auth_full()
|
|
users = cfg.get("credentials", {}).get("usernames", {})
|
|
if self.username in users:
|
|
users[self.username]["theme"] = value
|
|
_save_auth_full(cfg)
|
|
return self._apply_theme_script(value)
|
|
|
|
def handle_login(self, form_data: dict | None = None):
|
|
self.login_error = ""
|
|
users = self._load_users()
|
|
user = users.get(self.login_user)
|
|
if not user:
|
|
self.login_error = "Identifiant ou mot de passe incorrect"
|
|
self.login_pass = ""
|
|
return
|
|
try:
|
|
ok = bcrypt.checkpw(self.login_pass.encode(), user["password"].encode())
|
|
except Exception:
|
|
ok = False
|
|
if not ok:
|
|
self.login_error = "Identifiant ou mot de passe incorrect"
|
|
self.login_pass = ""
|
|
return
|
|
|
|
# Mot de passe OK — étape 2FA
|
|
self.login_pass = ""
|
|
self.totp_pending_user = self.login_user
|
|
self.totp_code = ""
|
|
self.totp_error = ""
|
|
|
|
if user.get("totp_secret"):
|
|
# 2FA déjà configurée — demander le code
|
|
self.totp_step = "verify"
|
|
self.totp_secret_pending = ""
|
|
self.totp_qr_data_url = ""
|
|
else:
|
|
# Première connexion 2FA — générer secret + QR
|
|
secret = pyotp.random_base32()
|
|
self.totp_secret_pending = secret
|
|
label = f"{self.login_user}@{TOTP_ISSUER}"
|
|
self.totp_qr_data_url = _make_totp_qr_data_url(secret, label)
|
|
self.totp_step = "setup"
|
|
|
|
def verify_totp(self, form_data: dict | None = None):
|
|
"""Vérifie le code TOTP saisi. Si setup, sauve le secret. Puis login."""
|
|
self.totp_error = ""
|
|
if not self.totp_pending_user:
|
|
# Session perdue — retour login
|
|
self.cancel_totp()
|
|
return rx.redirect("/login")
|
|
|
|
if len(self.totp_code) != 6:
|
|
self.totp_error = "Code à 6 chiffres requis"
|
|
return
|
|
|
|
# Déterminer le secret à valider
|
|
if self.totp_step == "setup":
|
|
secret = self.totp_secret_pending
|
|
else:
|
|
cfg = _load_auth_full()
|
|
user = cfg.get("credentials", {}).get("usernames", {}).get(self.totp_pending_user)
|
|
if not user:
|
|
self.cancel_totp()
|
|
return rx.redirect("/login")
|
|
secret = user.get("totp_secret") or ""
|
|
|
|
if not secret:
|
|
self.totp_error = "Configuration 2FA manquante — contactez un administrateur"
|
|
return
|
|
|
|
# Vérifier code (valid_window=1 → tolère ±30s de dérive d'horloge)
|
|
totp = pyotp.TOTP(secret)
|
|
if not totp.verify(self.totp_code, valid_window=1):
|
|
self.totp_error = "Code invalide"
|
|
self.totp_code = ""
|
|
return
|
|
|
|
# Code OK
|
|
cfg = _load_auth_full()
|
|
users = cfg.get("credentials", {}).get("usernames", {})
|
|
user = users.get(self.totp_pending_user)
|
|
if not user:
|
|
self.cancel_totp()
|
|
return rx.redirect("/login")
|
|
|
|
# Si setup, sauver le secret dans auth.yaml
|
|
if self.totp_step == "setup":
|
|
users[self.totp_pending_user]["totp_secret"] = secret
|
|
_save_auth_full(cfg)
|
|
|
|
# Finaliser la connexion
|
|
self.username = self.totp_pending_user
|
|
self.name = user.get("name", self.totp_pending_user)
|
|
self.role = user.get("role", "user")
|
|
self.photo_url = user.get("avatar_url", "")
|
|
self.theme = user.get("theme") or "eptm"
|
|
# Reset le flag de dismiss du popup d'enrôlement à chaque login —
|
|
# si l'user n'a toujours pas de classes, le popup doit ré-apparaître.
|
|
self.enroll_dismissed = False
|
|
self._reset_totp_flow()
|
|
return [self._apply_theme_script(self.theme), rx.redirect("/accueil")]
|
|
|
|
def cancel_totp(self):
|
|
"""Annule le flow 2FA et revient à l'étape password."""
|
|
self._reset_totp_flow()
|
|
|
|
def _reset_totp_flow(self):
|
|
self.totp_step = "password"
|
|
self.totp_pending_user = ""
|
|
self.totp_secret_pending = ""
|
|
self.totp_qr_data_url = ""
|
|
self.totp_code = ""
|
|
self.totp_error = ""
|
|
|
|
def logout(self):
|
|
self._clear_session()
|
|
return rx.redirect("/login")
|
|
|
|
def _clear_session(self):
|
|
self.username = ""
|
|
self.name = ""
|
|
self.role = "user"
|
|
self.photo_url = ""
|
|
self.theme = "eptm"
|
|
self.must_enroll = False
|
|
self.enroll_dismissed = False
|
|
self.my_classes = []
|
|
self.classes_unknown = []
|
|
self.escada_username = ""
|
|
self.escada_has_password = False
|
|
self.login_user = ""
|
|
self.login_pass = ""
|
|
self.login_error = ""
|
|
self._reset_totp_flow()
|
|
|
|
@staticmethod
|
|
def _load_users() -> dict:
|
|
return _load_auth_full().get("credentials", {}).get("usernames", {})
|