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