"""Auth flows: register, login, logout, email verification, password reset.""" from datetime import datetime from app.utils.time import utcnow from flask import (Blueprint, render_template, redirect, url_for, flash, request, current_app, abort) from flask_login import login_user, logout_user, login_required, current_user from flask_babel import gettext as _ from app.extensions import db, limiter from app.models.user import User from app.models.plan import Plan from app.models.enums import Role, TrustEventType from app.utils.security import generate_token, read_token from app.services.email import send_email from app.services.turnstile import verify_turnstile, turnstile_enabled from app.services.trust import record_event from app.services.settings import get_setting from app.blueprints.auth.forms import (RegisterForm, LoginForm, ResetRequestForm, ResetForm) auth_bp = Blueprint("auth", __name__, url_prefix="/auth") _VERIFY_SALT = "email-verify" _RESET_SALT = "password-reset" def _send_verify_email(user): token = generate_token(user.id, _VERIFY_SALT) link = url_for("auth.verify_email", token=token, _external=True) send_email(user.email, _("Verify your email"), _("Confirm your account: %(link)s", link=link)) @auth_bp.route("/register", methods=["GET", "POST"]) @limiter.limit("10 per hour", methods=["POST"]) def register(): if current_user.is_authenticated: return redirect(url_for("main.index")) form = RegisterForm() if form.validate_on_submit(): if not get_setting("registration_open", True): flash(_("Registration is currently closed."), "danger") return render_template("auth/register.html", form=form, turnstile=turnstile_enabled()) if not verify_turnstile(): flash(_("Captcha verification failed."), "danger") return render_template("auth/register.html", form=form, turnstile=turnstile_enabled()) existing = User.query.filter_by(email=form.email.data.lower()).first() if existing: flash(_("An account with that email already exists."), "danger") return render_template("auth/register.html", form=form, turnstile=turnstile_enabled()) free_plan = Plan.query.filter_by(slug="free").first() user = User( email=form.email.data.lower(), display_name=form.display_name.data.strip(), role=Role.free, tier_id=free_plan.id if free_plan else None, ) user.set_password(form.password.data) db.session.add(user) db.session.commit() _send_verify_email(user) flash(_("Account created. Check your email to verify."), "success") return redirect(url_for("auth.login")) return render_template("auth/register.html", form=form, turnstile=turnstile_enabled()) @auth_bp.route("/login", methods=["GET", "POST"]) @limiter.limit("20 per hour", methods=["POST"]) def login(): if current_user.is_authenticated: return redirect(url_for("main.index")) form = LoginForm() if form.validate_on_submit(): user = User.query.filter_by(email=form.email.data.lower()).first() if user is None or not user.check_password(form.password.data): flash(_("Invalid email or password."), "danger") return render_template("auth/login.html", form=form) if not user.is_active: flash(_("This account is suspended."), "danger") return render_template("auth/login.html", form=form) login_user(user, remember=form.remember.data) user.last_login_at = utcnow() db.session.commit() nxt = request.args.get("next") if nxt and nxt.startswith("/"): return redirect(nxt) return redirect(url_for("main.index")) return render_template("auth/login.html", form=form) @auth_bp.route("/logout") @login_required def logout(): logout_user() flash(_("Signed out."), "info") return redirect(url_for("main.index")) @auth_bp.route("/verify/") def verify_email(token): user_id = read_token(token, _VERIFY_SALT, current_app.config["TOKEN_VERIFY_MAX_AGE"]) if user_id is None: flash(_("Verification link is invalid or expired."), "danger") return redirect(url_for("auth.login")) user = db.session.get(User, user_id) if user is None: abort(404) if not user.email_verified: user.email_verified = True record_event(user, TrustEventType.verified_email, 5) db.session.commit() flash(_("Email verified. You're all set."), "success") return redirect(url_for("auth.login")) @auth_bp.route("/reset", methods=["GET", "POST"]) @limiter.limit("5 per hour", methods=["POST"]) def reset_request(): form = ResetRequestForm() if form.validate_on_submit(): user = User.query.filter_by(email=form.email.data.lower()).first() if user: token = generate_token(user.id, _RESET_SALT) link = url_for("auth.reset_password", token=token, _external=True) send_email(user.email, _("Reset your password"), _("Reset link: %(link)s", link=link)) # Always show success to avoid email enumeration. flash(_("If that email exists, a reset link has been sent."), "info") return redirect(url_for("auth.login")) return render_template("auth/reset_request.html", form=form) @auth_bp.route("/reset/", methods=["GET", "POST"]) def reset_password(token): user_id = read_token(token, _RESET_SALT, current_app.config["TOKEN_RESET_MAX_AGE"]) if user_id is None: flash(_("Reset link is invalid or expired."), "danger") return redirect(url_for("auth.reset_request")) user = db.session.get(User, user_id) if user is None: abort(404) form = ResetForm() if form.validate_on_submit(): user.set_password(form.password.data) db.session.commit() flash(_("Password updated. Sign in."), "success") return redirect(url_for("auth.login")) return render_template("auth/reset.html", form=form, token=token)