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/<token>")
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,
)