Spaces:
Sleeping
Sleeping
| """ | |
| Authentication utilities and decorators for Flask-Login integration | |
| """ | |
| import re | |
| from functools import wraps | |
| from flask import redirect, url_for, request, flash, current_app | |
| from flask_login import current_user, LoginManager | |
| from email_validator import validate_email, EmailNotValidError | |
| import logging | |
| from models import User | |
| logger = logging.getLogger(__name__) | |
| # Initialize Flask-Login | |
| login_manager = LoginManager() | |
| def init_login_manager(app): | |
| """Initialize Flask-Login with the Flask app""" | |
| login_manager.init_app(app) | |
| login_manager.login_view = 'login' | |
| login_manager.login_message = 'Please log in to access this page.' | |
| login_manager.login_message_category = 'info' | |
| def load_user(user_id): | |
| """User loader function for Flask-Login""" | |
| try: | |
| return User.find_by_id(user_id) | |
| except Exception as e: | |
| logger.error(f"Error loading user with ID {user_id}: {e}") | |
| return None | |
| def login_required(f): | |
| """ | |
| Decorator to require authentication for routes | |
| Enhanced version that handles AJAX requests properly | |
| """ | |
| def decorated_function(*args, **kwargs): | |
| if not current_user.is_authenticated: | |
| # Handle AJAX requests | |
| if request.is_json or request.headers.get('Content-Type') == 'application/json': | |
| from flask import jsonify | |
| return jsonify({ | |
| 'success': False, | |
| 'error': 'Authentication required', | |
| 'redirect': url_for('login') | |
| }), 401 | |
| # Handle regular requests | |
| flash('Please log in to access this page.', 'info') | |
| return redirect(url_for('login', next=request.url)) | |
| return f(*args, **kwargs) | |
| return decorated_function | |
| def get_current_user(): | |
| """Get currently logged in user""" | |
| try: | |
| if current_user.is_authenticated: | |
| return current_user | |
| return None | |
| except Exception as e: | |
| logger.error(f"Error getting current user: {e}") | |
| return None | |
| def validate_email_format(email: str) -> tuple[bool, str]: | |
| """ | |
| Validate email format using email-validator | |
| Returns (is_valid, error_message) | |
| """ | |
| try: | |
| if not email or not email.strip(): | |
| return False, "Email is required" | |
| email = email.strip() | |
| # Use email-validator library for comprehensive validation | |
| validate_email(email, check_deliverability=False) | |
| return True, "" | |
| except EmailNotValidError as e: | |
| return False, "Please enter a valid email address" | |
| except Exception as e: | |
| logger.error(f"Error validating email {email}: {e}") | |
| return False, "Email validation error" | |
| def validate_password_format(password: str) -> tuple[bool, str]: | |
| """ | |
| Validate password strength requirements | |
| Returns (is_valid, error_message) | |
| """ | |
| try: | |
| if not password: | |
| return False, "Password is required" | |
| # Check minimum length (8 characters as per requirements) | |
| if len(password) < 8: | |
| return False, "Password must be at least 8 characters long" | |
| # Additional strength checks can be added here if needed | |
| # For now, just checking minimum length as per requirements | |
| return True, "" | |
| except Exception as e: | |
| logger.error(f"Error validating password: {e}") | |
| return False, "Password validation error" | |
| def validate_registration_data(email: str, password: str, confirm_password: str) -> tuple[bool, list]: | |
| """ | |
| Validate complete registration data | |
| Returns (is_valid, list_of_errors) | |
| """ | |
| errors = [] | |
| try: | |
| # Validate email | |
| email_valid, email_error = validate_email_format(email) | |
| if not email_valid: | |
| errors.append(email_error) | |
| # Validate password | |
| password_valid, password_error = validate_password_format(password) | |
| if not password_valid: | |
| errors.append(password_error) | |
| # Check password confirmation | |
| if password != confirm_password: | |
| errors.append("Passwords do not match") | |
| # Check if email already exists (if email format is valid) | |
| if email_valid and User.find_by_email(email.strip()): | |
| errors.append("Email already registered") | |
| return len(errors) == 0, errors | |
| except Exception as e: | |
| logger.error(f"Error validating registration data: {e}") | |
| return False, ["Registration validation error"] | |
| def validate_login_data(email: str, password: str) -> tuple[bool, list]: | |
| """ | |
| Validate login data | |
| Returns (is_valid, list_of_errors) | |
| """ | |
| errors = [] | |
| try: | |
| # Basic validation | |
| if not email or not email.strip(): | |
| errors.append("Email is required") | |
| if not password: | |
| errors.append("Password is required") | |
| # If basic validation passes, check format | |
| if not errors: | |
| email_valid, email_error = validate_email_format(email) | |
| if not email_valid: | |
| errors.append("Please enter a valid email address") | |
| return len(errors) == 0, errors | |
| except Exception as e: | |
| logger.error(f"Error validating login data: {e}") | |
| return False, ["Login validation error"] | |
| def authenticate_user(email: str, password: str) -> tuple[bool, User, str]: | |
| """ | |
| Authenticate user with email and password | |
| Returns (is_authenticated, user_object, error_message) | |
| """ | |
| try: | |
| # Validate input data | |
| data_valid, errors = validate_login_data(email, password) | |
| if not data_valid: | |
| return False, None, errors[0] if errors else "Please check your email and password" | |
| # Find user by email | |
| user = User.find_by_email(email.strip()) | |
| if not user: | |
| logger.warning(f"Login attempt with non-existent email: {email}") | |
| return False, None, "Invalid email or password" | |
| # Check if user is active | |
| if not user.is_active: | |
| logger.warning(f"Login attempt for disabled account: {email}") | |
| return False, None, "Your account has been disabled. Please contact support." | |
| # Verify password | |
| if not user.check_password(password): | |
| logger.warning(f"Failed login attempt for user: {email}") | |
| return False, None, "Invalid email or password" | |
| logger.info(f"User authenticated successfully: {email}") | |
| return True, user, "" | |
| except Exception as e: | |
| logger.error(f"Error authenticating user {email}: {e}") | |
| return False, None, "We're experiencing technical difficulties. Please try again later." | |
| def create_user_account(email: str, password: str, confirm_password: str) -> tuple[bool, User, list]: | |
| """ | |
| Create new user account with validation | |
| Returns (is_created, user_object, list_of_errors) | |
| """ | |
| try: | |
| # Validate registration data | |
| data_valid, errors = validate_registration_data(email, password, confirm_password) | |
| if not data_valid: | |
| return False, None, errors | |
| # Create user | |
| user = User.create_user(email.strip(), password) | |
| if not user: | |
| logger.error(f"Failed to create user account for {email}") | |
| return False, None, ["We couldn't create your account. Please try again later."] | |
| logger.info(f"User account created successfully: {email}") | |
| return True, user, [] | |
| except Exception as e: | |
| logger.error(f"Error creating user account for {email}: {e}") | |
| return False, None, ["We're experiencing technical difficulties. Please try again later."] | |
| def is_safe_url(target): | |
| """ | |
| Check if a URL is safe for redirects | |
| Prevents open redirect vulnerabilities | |
| """ | |
| try: | |
| from urllib.parse import urlparse, urljoin | |
| from flask import request | |
| ref_url = urlparse(request.host_url) | |
| test_url = urlparse(urljoin(request.host_url, target)) | |
| return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc | |
| except Exception as e: | |
| logger.error(f"Error checking URL safety for {target}: {e}") | |
| return False | |
| def get_redirect_target(): | |
| """ | |
| Get safe redirect target from request | |
| """ | |
| try: | |
| for target in request.values.get('next'), request.referrer: | |
| if not target: | |
| continue | |
| if is_safe_url(target): | |
| return target | |
| return None | |
| except Exception as e: | |
| logger.error(f"Error getting redirect target: {e}") | |
| return None | |
| def redirect_back(endpoint, **values): | |
| """ | |
| Redirect back to a safe URL or fallback to endpoint | |
| """ | |
| try: | |
| target = get_redirect_target() | |
| if target: | |
| return redirect(target) | |
| return redirect(url_for(endpoint, **values)) | |
| except Exception as e: | |
| logger.error(f"Error in redirect_back: {e}") | |
| return redirect(url_for(endpoint, **values)) |