265 lines
9.3 KiB
Python
265 lines
9.3 KiB
Python
"""Authentication blueprint - login, logout, profile."""
|
|
from functools import wraps
|
|
|
|
from flask import Blueprint, abort, flash, jsonify, redirect, render_template, request, session, url_for
|
|
from flask_babel import gettext as _
|
|
|
|
from services.api_client import api_client
|
|
|
|
auth_bp = Blueprint("auth", __name__, url_prefix="/auth")
|
|
|
|
|
|
def login_required(f):
|
|
"""Decorator to require login for protected routes."""
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if not session.get("api_key"):
|
|
flash(_("Effettua il login per continuare"), "warning")
|
|
return redirect(url_for("auth.login"))
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
def role_required(*roles):
|
|
"""Decorator to require one of the specified roles.
|
|
|
|
Usage:
|
|
@role_required("MeasurementTec", "Metrologist")
|
|
def my_view(): ...
|
|
"""
|
|
def decorator(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
user = session.get("user", {})
|
|
user_roles = user.get("roles", [])
|
|
if not any(r in user_roles for r in roles):
|
|
abort(403)
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
return decorator
|
|
|
|
|
|
def _post_login_landing():
|
|
"""Pick the landing endpoint based on the logged-in user's roles.
|
|
|
|
A Maker-only user must not be bounced to /measure (MeasurementTec-only),
|
|
which would raise a 403 right after login. Order of preference follows the
|
|
primary operator flow first, then editing, analysis, admin.
|
|
"""
|
|
user = session.get("user", {})
|
|
roles = user.get("roles", [])
|
|
if "MeasurementTec" in roles:
|
|
return url_for("measure.select_recipe")
|
|
if "Maker" in roles:
|
|
return url_for("maker.recipe_list")
|
|
if "Metrologist" in roles:
|
|
return url_for("statistics.dashboard")
|
|
if user.get("is_admin"):
|
|
return url_for("admin.user_list")
|
|
return url_for("auth.profile")
|
|
|
|
|
|
def _complete_login(response, fallback_name):
|
|
"""Dopo una risposta di login riuscita: sessione, preferenze, impostazioni (#59).
|
|
|
|
Comune al login con password e al login con PIN, cosi' i due restano identici.
|
|
Ritorna l'utente salvato in sessione.
|
|
"""
|
|
# Salva in sessione. permanent=True fa valere PERMANENT_SESSION_LIFETIME
|
|
# (8h, config.py): senza, Flask usa una sessione di durata browser e il
|
|
# limite configurato non ha mai effetto (#33).
|
|
session.permanent = True
|
|
session["api_key"] = response["api_key"]
|
|
session["user"] = response["user"]
|
|
session["user_id"] = response["user"]["id"]
|
|
|
|
# Applica preferenze utente
|
|
user = response["user"]
|
|
if user.get("language_pref"):
|
|
session["language"] = user["language_pref"]
|
|
if user.get("theme_pref"):
|
|
session["theme"] = user["theme_pref"]
|
|
|
|
# Carica impostazioni di sistema
|
|
settings_resp = api_client.get("/api/settings")
|
|
if not settings_resp.get("error"):
|
|
logo_path = settings_resp.get("company_logo_path")
|
|
if logo_path:
|
|
session["company_logo"] = logo_path
|
|
auto_logout = settings_resp.get("auto_logout_minutes")
|
|
if auto_logout:
|
|
try:
|
|
session["auto_logout_minutes"] = int(auto_logout)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
flash(_("Benvenuto, %(name)s!", name=user.get("display_name", fallback_name)), "success")
|
|
|
|
return user
|
|
|
|
|
|
@auth_bp.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
"""Login page and form handler."""
|
|
# Se già loggato, redirect alla home in base al ruolo
|
|
if session.get("api_key"):
|
|
return redirect(_post_login_landing())
|
|
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
|
|
if not username or not password:
|
|
flash(_("Inserisci username e password"), "error")
|
|
return render_template("auth/login.html")
|
|
|
|
try:
|
|
response = api_client.post("/api/auth/login", data={"username": username, "password": password})
|
|
|
|
if response.get("error"):
|
|
flash(response.get("detail", _("Credenziali non valide")), "error")
|
|
return render_template("auth/login.html")
|
|
|
|
user = _complete_login(response, username)
|
|
|
|
# Redirect dopo login (in base al ruolo)
|
|
return redirect(_post_login_landing())
|
|
|
|
except Exception as e:
|
|
flash(_("Errore di connessione al server: %(error)s", error=str(e)), "error")
|
|
return render_template("auth/login.html")
|
|
|
|
return render_template("auth/login.html")
|
|
|
|
|
|
@auth_bp.route("/pin-users", methods=["GET"])
|
|
def pin_users():
|
|
"""Elenco nomi per la vista Operatore del login (#59): solo id e nome.
|
|
|
|
Pubblica come il login; se il backend non risponde o non ha il PIN, lista
|
|
vuota e la pagina apre il form password.
|
|
"""
|
|
if session.get("api_key"):
|
|
return jsonify([])
|
|
try:
|
|
resp = api_client.get("/api/auth/pin-users")
|
|
except Exception:
|
|
return jsonify([])
|
|
if not isinstance(resp, list):
|
|
return jsonify([])
|
|
return jsonify([
|
|
{"id": u.get("id"), "display_name": u.get("display_name", "")}
|
|
for u in resp if isinstance(u, dict)
|
|
])
|
|
|
|
|
|
@auth_bp.route("/login-pin", methods=["POST"])
|
|
def login_pin():
|
|
"""Login con nome + PIN (#59). JSON in/out; CSRF come il login password
|
|
(header X-CSRFToken, gestito da CSRFProtect)."""
|
|
data = request.get_json(silent=True) or {}
|
|
user_id = data.get("user_id")
|
|
pin = str(data.get("pin", ""))
|
|
if not isinstance(user_id, int) or not (len(pin) == 4 and pin.isdigit()):
|
|
return jsonify({"error": True, "status_code": 422, "detail": _("Il PIN deve essere di 4 cifre")}), 422
|
|
|
|
try:
|
|
response = api_client.post("/api/auth/login-pin", data={"user_id": user_id, "pin": pin})
|
|
except Exception as e:
|
|
return jsonify({"error": True, "status_code": 502,
|
|
"detail": _("Errore di connessione al server: %(error)s", error=str(e))}), 502
|
|
|
|
if response.get("error"):
|
|
status = response.get("status_code") or 401
|
|
if status == 401:
|
|
detail = _("PIN errato")
|
|
else:
|
|
detail = response.get("detail") or _("Credenziali non valide")
|
|
return jsonify({"error": True, "status_code": status, "detail": detail}), status
|
|
|
|
_complete_login(response, "") # include il flash di benvenuto
|
|
return jsonify({"ok": True, "redirect": _post_login_landing()})
|
|
|
|
|
|
@auth_bp.route("/logout", methods=["POST"])
|
|
def logout():
|
|
"""Clear session and redirect to login."""
|
|
# Invalida il token sul server
|
|
if session.get("api_key"):
|
|
try:
|
|
api_client.post("/api/auth/logout")
|
|
except Exception:
|
|
pass # Ignora errori durante logout
|
|
|
|
session.clear()
|
|
flash(_("Logout effettuato"), "info")
|
|
return redirect(url_for("auth.login"))
|
|
|
|
|
|
@auth_bp.route("/set-theme", methods=["POST"])
|
|
@login_required
|
|
def set_theme():
|
|
"""Persist the user's theme toggle (light/dark) to session and profile.
|
|
|
|
Called via AJAX from the navbar theme switch so the choice survives
|
|
navigation and matches on the next login. Best-effort on the API side:
|
|
the session is updated regardless so the current browsing stays correct.
|
|
"""
|
|
theme = (request.get_json(silent=True) or {}).get("theme", "")
|
|
if theme not in ("light", "dark"):
|
|
return jsonify({"error": True, "detail": "invalid theme"}), 400
|
|
|
|
session["theme"] = theme
|
|
try:
|
|
api_client.put("/api/auth/me", data={"theme_pref": theme})
|
|
except Exception:
|
|
pass # Non-fatal: session theme is already updated for this session.
|
|
return jsonify({"ok": True, "theme": theme})
|
|
|
|
|
|
@auth_bp.route("/profile", methods=["GET", "POST"])
|
|
@login_required
|
|
def profile():
|
|
"""User profile page - change display name, language, theme."""
|
|
if request.method == "POST":
|
|
display_name = request.form.get("display_name", "").strip()
|
|
language_pref = request.form.get("language_pref", "").strip()
|
|
theme_pref = request.form.get("theme_pref", "").strip()
|
|
|
|
try:
|
|
data = {}
|
|
if display_name:
|
|
data["display_name"] = display_name
|
|
if language_pref:
|
|
data["language_pref"] = language_pref
|
|
if theme_pref:
|
|
data["theme_pref"] = theme_pref
|
|
|
|
response = api_client.put("/api/auth/me", data=data)
|
|
|
|
if response.get("error"):
|
|
flash(response.get("detail", _("Errore durante l'aggiornamento del profilo")), "error")
|
|
else:
|
|
# Aggiorna sessione
|
|
session["user"] = response
|
|
if language_pref:
|
|
session["language"] = language_pref
|
|
if theme_pref:
|
|
session["theme"] = theme_pref
|
|
|
|
flash(_("Profilo aggiornato con successo"), "success")
|
|
return redirect(url_for("auth.profile"))
|
|
|
|
except Exception as e:
|
|
flash(_("Errore di connessione al server: %(error)s", error=str(e)), "error")
|
|
|
|
# GET - carica dati profilo
|
|
user_data = api_client.get("/api/auth/me")
|
|
if user_data.get("error"):
|
|
flash(_("Errore nel caricamento del profilo: %(error)s", error=user_data.get("detail", "")), "error")
|
|
user_data = session.get("user", {})
|
|
else:
|
|
session["user"] = user_data
|
|
|
|
return render_template("auth/profile.html", user=user_data)
|