# auth.py - Complete authentication system from flask import Blueprint, render_template, request, redirect, url_for, flash, session from flask_login import LoginManager, login_user, logout_user, login_required, current_user from werkzeug.security import generate_password_hash, check_password_hash from models import db, User, Organization import secrets import re auth_bp = Blueprint('auth', __name__) # Email validation def is_valid_email(email): pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None # Password validation def is_strong_password(password): if len(password) < 8: return False, "Password must be at least 8 characters long" if not re.search(r'[A-Z]', password): return False, "Password must contain at least one uppercase letter" if not re.search(r'[a-z]', password): return False, "Password must contain at least one lowercase letter" if not re.search(r'[0-9]', password): return False, "Password must contain at least one number" return True, "Password is strong" @auth_bp.route('/login', methods=['GET', 'POST']) def login(): if current_user.is_authenticated: return redirect(url_for('dashboard')) if request.method == 'POST': email = request.form.get('email', '').strip().lower() password = request.form.get('password', '') remember = request.form.get('remember') == 'on' # Validation if not email or not password: flash('Please fill in all fields', 'error') return render_template('auth/login.html') if not is_valid_email(email): flash('Please enter a valid email address', 'error') return render_template('auth/login.html') # Find user user = User.query.filter_by(email=email, is_active=True).first() if user and check_password_hash(user.password_hash, password): # Check if organization subscription is active if user.organization.subscription_status != 'active': flash('Your organization subscription has expired. Please contact support.', 'warning') return redirect(url_for('billing.billing')) login_user(user, remember=remember) user.last_login_at = datetime.utcnow() db.session.commit() # Redirect to next page or dashboard next_page = request.args.get('next') if next_page and url_parse(next_page).netloc == '': return redirect(next_page) return redirect(url_for('dashboard.index')) else: flash('Invalid email or password', 'error') return render_template('auth/login.html') @auth_bp.route('/register', methods=['GET', 'POST']) def register(): if current_user.is_authenticated: return redirect(url_for('dashboard.index')) if request.method == 'POST': # Get form data first_name = request.form.get('firstName', '').strip() last_name = request.form.get('lastName', '').strip() email = request.form.get('email', '').strip().lower() password = request.form.get('password', '') org_name = request.form.get('orgName', '').strip() org_size = request.form.get('orgSize', '') industry = request.form.get('industry', '') plan = request.form.get('plan', 'free') # Validation errors = [] if not all([first_name, last_name, email, password, org_name, org_size, industry]): errors.append('Please fill in all required fields') if not is_valid_email(email): errors.append('Please enter a valid email address') is_strong, password_msg = is_strong_password(password) if not is_strong: errors.append(password_msg) # Check if user already exists existing_user = User.query.filter_by(email=email).first() if existing_user: errors.append('An account with this email already exists') if errors: for error in errors: flash(error, 'error') return render_template('auth/register.html') try: # Create organization organization = Organization( name=org_name, size=org_size, industry=industry, subscription_plan=plan ) db.session.add(organization) db.session.flush() # Get the organization ID # Create user user = User( first_name=first_name, last_name=last_name, email=email, password_hash=generate_password_hash(password), organization_id=organization.id, role='admin', # First user is always admin verification_token=secrets.token_urlsafe(32) ) db.session.add(user) db.session.commit() # Send verification email (implement later) # send_verification_email(user) flash('Account created successfully! Please verify your email.', 'success') # Auto-login for now (disable in production until email verification) login_user(user) return redirect(url_for('onboarding.start')) except Exception as e: db.session.rollback() flash('An error occurred while creating your account. Please try again.', 'error') app.logger.error(f'Registration error: {str(e)}') return render_template('auth/register.html') @auth_bp.route('/logout') @login_required def logout(): logout_user() flash('You have been logged out successfully', 'info') return redirect(url_for('auth.login')) @auth_bp.route('/forgot-password', methods=['GET', 'POST']) def forgot_password(): if request.method == 'POST': email = request.form.get('email', '').strip().lower() if not is_valid_email(email): flash('Please enter a valid email address', 'error') return render_template('auth/forgot_password.html') user = User.query.filter_by(email=email, is_active=True).first() if user: # Generate reset token reset_token = secrets.token_urlsafe(32) user.reset_token = reset_token user.reset_token_expires = datetime.utcnow() + timedelta(hours=1) db.session.commit() # Send reset email (implement later) # send_password_reset_email(user, reset_token) # Always show success message for security flash('If an account with that email exists, we\'ve sent a password reset link.', 'info') return redirect(url_for('auth.login')) return render_template('auth/forgot_password.html') @auth_bp.route('/reset-password/', methods=['GET', 'POST']) def reset_password(token): user = User.query.filter_by( reset_token=token, is_active=True ).first() if not user or user.reset_token_expires < datetime.utcnow(): flash('Invalid or expired reset token', 'error') return redirect(url_for('auth.forgot_password')) if request.method == 'POST': password = request.form.get('password', '') confirm_password = request.form.get('confirmPassword', '') if password != confirm_password: flash('Passwords do not match', 'error') return render_template('auth/reset_password.html', token=token) is_strong, password_msg = is_strong_password(password) if not is_strong: flash(password_msg, 'error') return render_template('auth/reset_password.html', token=token) # Update password user.password_hash = generate_password_hash(password) user.reset_token = None user.reset_token_expires = None db.session.commit() flash('Your password has been updated successfully', 'success') return redirect(url_for('auth.login')) return render_template('auth/reset_password.html', token=token) # Initialize login manager def init_auth(app): login_manager = LoginManager() login_manager.init_app(app) login_manager.login_view = 'auth.login' login_manager.login_message = 'Please log in to access this page.' login_manager.login_message_category = 'info' @login_manager.user_loader def load_user(user_id): return User.query.get(int(user_id)) app.register_blueprint(auth_bp, url_prefix='/auth')