from functools import wraps from flask import Blueprint, abort, flash, jsonify, redirect, render_template, request, session, url_for from app.auth_utils import login_required from app.models import ( authenticate_user, generate_magic_login_token, generate_referral_code, get_user, is_magic_login_enabled, register_user, validate_magic_login_token, ) def current_user(): """Return current user dict from session, or None. Note: password_hash is NOT included — callers needing it for decryption should use _get_site_owner_password_hash(site_id) which fetches from DB. """ from app.models import get_user if "user_id" not in session: return None user = get_user(session["user_id"]) return user def verified_email_required(f): """Decorator that blocks unverified users from key actions. Unverified users can still log in and access their dashboard, but cannot create sites, generate API keys, or submit forms. """ @wraps(f) def decorated_function(*args, **kwargs): user = current_user() if user is None: return redirect(url_for("auth.login")) if not user.get("email_verified"): if request.is_json or request.accept_mimetypes.best == "application/json": return jsonify( { "error": "Email verification required", "message": "Please verify your email before performing this action.", } ), 403 flash("Please verify your email before performing this action.", "warning") return redirect(url_for("email.send_verification")) return f(*args, **kwargs) return decorated_function auth_bp = Blueprint("auth", __name__, url_prefix="/auth") @auth_bp.route("/register", methods=["GET", "POST"]) def register(): """User registration.""" if "user_id" in session: return redirect(url_for("site.index")) if request.method == "POST": email = request.form.get("email", "").strip().lower() password = request.form.get("password", "") name = request.form.get("name", "").strip() errors = [] if not email or "@" not in email: errors.append("Valid email required") if len(password) < 8: errors.append("Password must be at least 8 characters") if name and len(name) > 100: errors.append("Name too long") if errors: return render_template("register.html", errors=errors, email=email, name=name), 400 try: magic_login_enabled = request.form.get("magic_login_enabled") == "on" user = register_user(email, password, name, magic_login_enabled=magic_login_enabled) except ValueError as e: return render_template("register.html", errors=[str(e)], email=email, name=name), 409 # Auto-login after registration (C1: never store password_hash in session) # C3: session.clear() + new session data = fresh signed cookie (Flask client-side sessions) # Unlike server-side sessions, Flask's signed cookies don't need regenerate() — # the old cookie with cleared data is cryptographically invalidated. session.clear() session["user_id"] = user["id"] session.permanent = False # Fresh session, not "remember me" flash("Account created. Welcome to AgentForms!", "success") return redirect(url_for("auth.dashboard")) return render_template("register.html", errors=[]) @auth_bp.route("/signup") def signup_redirect(): """Redirect /auth/signup -> /auth/register (URL consistency).""" return redirect(url_for("auth.register")) @auth_bp.route("/login", methods=["GET", "POST"]) def login(): """User login.""" if "user_id" in session: return redirect(url_for("site.index")) next_url = request.args.get("next") if next_url: # Prevent open redirect — only allow relative paths if not next_url.startswith("/"): next_url = None if request.method == "POST": # Support both form data and JSON body if request.is_json: data = request.get_json(silent=True) or {} email = data.get("email", "").strip().lower() password = data.get("password", "") remember = bool(data.get("remember", False)) else: email = request.form.get("email", "").strip().lower() password = request.form.get("password", "") remember = request.form.get("remember") == "on" if not email or not password: if request.is_json: return jsonify({"error": "Email and password required"}), 400 return render_template( "login.html", errors=["Email and password required"], email=email, next=next_url ), 400 user = authenticate_user(email, password) if user is None: if request.is_json: return jsonify({"error": "Invalid email or password"}), 401 return render_template("login.html", errors=["Invalid email or password"], email=email, next=next_url), 401 # C3: session.clear() + new session data = fresh signed cookie (Flask client-side sessions) # The old signed cookie is cryptographically invalidated when content changes. session.clear() session["user_id"] = user["id"] # C1: password_hash no longer stored in session — fetched from DB when needed session.permanent = remember redirect_url = next_url or url_for("auth.dashboard") return redirect(redirect_url) return render_template("login.html", errors=[], next=next_url) @auth_bp.route("/logout", methods=["POST"]) def logout(): """User logout.""" session.clear() flash("Logged out.", "info") return redirect(url_for("site.index")) @auth_bp.route("/magic", methods=["POST"]) def magic_login_request(): """Request a magic link via email. JSON body: email: user email Also works as form POST for the login page flow. """ from app.routes.email import NotificationService if "user_id" in session: return redirect(url_for("site.index")) email = "" if request.is_json: data = request.get_json(silent=True) or {} email = data.get("email", "").strip().lower() else: email = request.form.get("email", "").strip().lower() if not email or "@" not in email: if request.is_json: from flask import jsonify return jsonify({"error": "Valid email is required"}), 400 return render_template("login.html", errors=["Valid email is required"], email=email, magic_error=True) token = generate_magic_login_token(email) # Check if magic login is enabled for this user magic_disabled = False if token: from app.models import find_user_by_email user = find_user_by_email(email) if user and not is_magic_login_enabled(user["id"]): token = None magic_disabled = True # Always return success to avoid email enumeration if token: base_url = request.host_url.rstrip("/") magic_link = f"{base_url}/auth/magic/{token}" body = f"""Hello, Click the link below to sign in to AgentForms: {magic_link} This link expires in 15 minutes and can only be used once. If you didn't request this, you can safely ignore this email. — AgentForms """ NotificationService.send(email, "Sign in to AgentForms", body) if request.is_json: from flask import jsonify if magic_disabled: return jsonify({"error": "Passwordless login is disabled for this account. Please use your password."}), 403 return jsonify( {"success": True, "message": "If an account with that email exists, you'll receive a login link."} ) if magic_disabled: return render_template( "login.html", errors=["Passwordless login is disabled for this account. Please use your password."], email=email, magic_error=True, ) flash("If an account with that email exists, you'll receive a login link.", "success") return redirect(url_for("auth.login")) @auth_bp.route("/magic/") def magic_login(token): """Consume a magic login token and create a session.""" if "user_id" in session: return redirect(url_for("site.index")) user_id = validate_magic_login_token(token) if user_id: # C3: session.clear() + new session data = fresh signed cookie (Flask client-side sessions) session.clear() session["user_id"] = user_id # C1: password_hash no longer stored in session user = get_user(user_id) flash(f"Welcome back, {user.get('name', 'there')}!", "success") return redirect(url_for("auth.dashboard")) flash("Invalid or expired login link. Please request a new one.", "error") return redirect(url_for("auth.login")) @auth_bp.route("/settings") @auth_bp.route("/dashboard") @login_required def dashboard(): """User dashboard — overview of account, sites, usage.""" from app.helpers import _get_site_owner_password_hash from app.models import get_user_site_count, get_user_sites, get_user_usage, list_api_keys, list_submissions user = current_user() sites = get_user_sites(user["id"]) site_count = get_user_site_count(user["id"]) usage = get_user_usage(user["id"], user["tier"]) api_key_count = len(list_api_keys(user["id"])) referral_code = generate_referral_code(user["id"]) # Recent submissions across user's sites (decrypted via DB lookup — C1 fix) site_ids = [s["id"] for s in sites] recent = [] if site_ids: for sid in site_ids: pw_hash = _get_site_owner_password_hash(sid) site_subs = list_submissions(limit=10, site_filter=sid, password_hash=pw_hash) recent.extend(site_subs) recent.sort(key=lambda x: x.get("submitted_at") or "", reverse=True) recent = recent[:10] return render_template( "user_dashboard.html", user=user, sites=sites, site_count=site_count, usage=usage, recent_submissions=recent, api_key_count=api_key_count, referral_code=referral_code, )