from flask import Blueprint, request, render_template, redirect, url_for, flash, session, jsonify, current_app from werkzeug.security import generate_password_hash, check_password_hash from functools import wraps from datetime import datetime, timedelta import uuid from database import db_session from models import User, Company, Session, Invitation, ActivityLog, PasswordReset from extensions import limiter # Rate limiting from security_utils import is_ip_blocked, record_failed_attempt # Brute-force protection # Create a Blueprint for auth routes auth_bp = Blueprint('auth', __name__) # Decorator functions for role-based access control def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'user_id' not in session: # Check if this is an API/AJAX request expecting JSON if request.is_json or request.path.startswith('/api/') or request.headers.get('Accept') == 'application/json': return jsonify({'error': 'Authentication required', 'redirect': url_for('auth.login')}), 401 return redirect(url_for('auth.login', next=request.url)) return f(*args, **kwargs) return decorated_function def admin_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'user_id' not in session: # Check if this is an API/AJAX request expecting JSON if request.is_json or request.path.startswith('/api/') or request.headers.get('Accept') == 'application/json': return jsonify({'error': 'Authentication required', 'redirect': url_for('auth.login')}), 401 return redirect(url_for('auth.login', next=request.url)) user = User.query.get(session['user_id']) if not user or not user.is_admin(): # Check if this is an API/AJAX request expecting JSON if request.is_json or request.path.startswith('/api/') or request.headers.get('Accept') == 'application/json': return jsonify({'error': 'Admin permission required'}), 403 flash('You do not have permission to access this page.', 'danger') return redirect(url_for('index')) return f(*args, **kwargs) return decorated_function def company_admin_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'user_id' not in session: # Check if this is an API/AJAX request expecting JSON if request.is_json or request.path.startswith('/api/') or request.headers.get('Accept') == 'application/json': return jsonify({'error': 'Authentication required', 'redirect': url_for('auth.login')}), 401 return redirect(url_for('auth.login', next=request.url)) user = User.query.get(session['user_id']) if not user or (not user.is_admin() and not user.is_company_admin()): # Check if this is an API/AJAX request expecting JSON if request.is_json or request.path.startswith('/api/') or request.headers.get('Accept') == 'application/json': return jsonify({'error': 'Company admin permission required'}), 403 flash('You do not have permission to access this page.', 'danger') return redirect(url_for('index')) return f(*args, **kwargs) return decorated_function # Function to log user activity def log_activity(user_id, activity_type, description, entity_type=None, entity_id=None): user = User.query.get(user_id) if not user: return activity = ActivityLog( user_id=user_id, company_id=user.company_id, activity_type=activity_type, description=description, entity_type=entity_type, entity_id=entity_id ) db_session.add(activity) db_session.commit() # Route for login page @auth_bp.route('/login', methods=['GET', 'POST']) @limiter.limit("10 per minute") # IP-based rate limit: max 10 login attempts per minute def login(): if request.method == 'POST': email = request.form.get('email') password = request.form.get('password') remember = 'remember' in request.form ip_address = request.remote_addr # Check if IP is blocked due to too many failed attempts if is_ip_blocked(ip_address): flash('Too many failed login attempts from your IP address. Please try again later.', 'danger') return render_template('login.html'), 429 user = User.query.filter_by(email=email).first() # Check if user exists and password is correct if user and user.check_password(password): # Check if account is locked if user.locked_until and user.locked_until > datetime.utcnow(): minutes_left = int((user.locked_until - datetime.utcnow()).total_seconds() / 60) + 1 flash(f'Account temporarily locked due to too many failed login attempts. Try again in {minutes_left} minutes.', 'warning') return render_template('login.html'), 423 # Check if email is verified (if required) if not user.email_verified and hasattr(user, 'email_verified'): flash('Please verify your email address before logging in. Check your inbox for the verification link.', 'warning') return render_template('login.html') # Check if account is active if not user.active: flash('Your account has been deactivated. Please contact support.', 'danger') return render_template('login.html') if user and user.check_password(password) and user.active: # Reset failed login attempts on successful login user.failed_login_attempts = 0 user.locked_until = None # Update last login time user.last_login = datetime.utcnow() # Create a new session days_valid = 30 if remember else 1 user_session = Session.create_for_user(user.id, days_valid=days_valid) db_session.add(user_session) db_session.commit() # Set session variables session['user_id'] = user.id session['session_token'] = user_session.session_token # Log the activity log_activity(user.id, 'login', f"User logged in: {user.email}") # Redirect next_url = request.args.get('next') if next_url: return redirect(next_url) elif user.is_admin(): return redirect(url_for('admin.dashboard')) else: return redirect(url_for('dashboard')) else: # Failed login - record attempt and handle account lockout record_failed_attempt(ip_address, email) if user: # Increment failed login counter for this user user.failed_login_attempts = (user.failed_login_attempts or 0) + 1 # Check if we need to lock the account max_attempts = current_app.config.get('MAX_LOGIN_ATTEMPTS', 5) if user.failed_login_attempts >= max_attempts: # Lock the account lockout_minutes = current_app.config.get('LOCKOUT_DURATION_MINUTES', 30) user.locked_until = datetime.utcnow() + timedelta(minutes=lockout_minutes) db_session.commit() flash(f'Account locked due to too many failed login attempts. Try again in {lockout_minutes} minutes.', 'danger') else: # Show remaining attempts remaining = max_attempts - user.failed_login_attempts db_session.commit() flash(f'Invalid email or password. {remaining} attempt(s) remaining before account lockout.', 'danger') else: # User doesn't exist, but don't reveal that flash('Invalid email or password', 'danger') return render_template('login.html') # Route for logout @auth_bp.route('/logout') def logout(): if 'user_id' in session: user_id = session['user_id'] session_token = session.get('session_token') # Delete the session from the database if session_token: db_session.query(Session).filter_by(session_token=session_token).delete() db_session.commit() # Log the activity log_activity(user_id, 'logout', f"User logged out") # Clear session session.clear() return redirect(url_for('auth.login')) # Route for registration (for system admin to create the first admin) @auth_bp.route('/register', methods=['GET', 'POST']) def register(): # Check if there are any users in the system user_count = User.query.count() # If users exist, redirect non-admins if user_count > 0 and 'user_id' in session: user = User.query.get(session['user_id']) if not user or not user.is_admin(): flash('Only administrators can register new users', 'danger') return redirect(url_for('index')) if request.method == 'POST': email = request.form.get('email') password = request.form.get('password') confirm_password = request.form.get('confirm_password') first_name = request.form.get('first_name') last_name = request.form.get('last_name') phone_number = request.form.get('phone_number') # Validation - All fields are now required if not all([email, password, confirm_password, first_name, last_name, phone_number]): flash('All fields are required', 'danger') return render_template('auth/register.html') if password != confirm_password: flash('Passwords do not match', 'danger') return render_template('auth/register.html') # Check if email already exists (no more username check) if User.query.filter_by(email=email).first(): flash('Email already exists', 'danger') return render_template('auth/register.html') # Create new user with email as username # If this is the first user, make them an admin role = 'admin' if user_count == 0 else 'user' user = User( username=email, # ✅ SET USERNAME = EMAIL email=email, first_name=first_name, last_name=last_name, phone_number=phone_number, role=role ) user.set_password(password) db_session.add(user) db_session.commit() # Send registration notification email to admin try: from email_config import send_registration_notification send_registration_notification(user) except Exception as e: current_app.logger.error(f"Failed to send registration notification: {str(e)}") flash('Registration successful! You can now log in with your email address.', 'success') return redirect(url_for('auth.login')) return render_template('auth/register.html') # Route for accepting invitations @auth_bp.route('/accept-invitation/', methods=['GET', 'POST']) def accept_invitation(token): # Find the invitation invitation = Invitation.query.filter_by(invitation_token=token, accepted=False).first() if not invitation or invitation.is_expired: flash('Invalid or expired invitation', 'danger') return redirect(url_for('auth.login')) if request.method == 'POST': password = request.form.get('password') confirm_password = request.form.get('confirm_password') first_name = request.form.get('first_name') last_name = request.form.get('last_name') phone_number = request.form.get('phone_number') # Validation - All fields are now required if not all([password, confirm_password, first_name, last_name, phone_number]): flash('All fields are required', 'danger') return render_template('auth/accept_invitation.html', invitation=invitation) if password != confirm_password: flash('Passwords do not match', 'danger') return render_template('auth/accept_invitation.html', invitation=invitation) # Use email as username - check if user already exists by email if User.query.filter_by(email=invitation.email).first(): flash('A user with this email already exists', 'danger') return render_template('auth/accept_invitation.html', invitation=invitation) # Create new user with email as username user = User( username=invitation.email, # ✅ SET USERNAME = EMAIL email=invitation.email, first_name=first_name, last_name=last_name, phone_number=phone_number, role=invitation.role, company_id=invitation.company_id ) user.set_password(password) # Mark invitation as accepted invitation.accepted = True db_session.add(user) db_session.commit() # Log the activity log_activity( invitation.created_by, 'user_creation', f"New user registered via invitation: {user.email}" # Log email instead of username ) flash('Registration successful! You can now log in with your email address.', 'success') return redirect(url_for('auth.login')) return render_template('auth/accept_invitation.html', invitation=invitation) # Route for password reset request @auth_bp.route('/reset-password-request', methods=['GET', 'POST']) def reset_password_request(): if request.method == 'POST': email = request.form.get('email') user = User.query.filter_by(email=email).first() if user: # Check if there's already a valid reset token for this user existing_reset = PasswordReset.query.filter_by( user_id=user.id, used=False ).filter(PasswordReset.expires_at > datetime.utcnow()).first() if existing_reset: # Delete the existing token db_session.delete(existing_reset) db_session.commit() # Create a new password reset token reset_token = PasswordReset.create_token(user.id) db_session.add(reset_token) db_session.commit() # Send email with reset link reset_url = url_for( 'auth.reset_password', token=reset_token.token, _external=True ) # Try to send email try: from email_config import send_password_reset_email if send_password_reset_email(user.email, reset_url): flash('If your email is registered, you will receive a password reset link', 'info') else: # Email failed, show the link in flash message for testing flash(f'Email sending failed. Reset link: {reset_url}', 'warning') print(f"Password reset URL for {user.email}: {reset_url}") except Exception as e: # Fallback: show the link flash(f'Email system error. Reset link: {reset_url}', 'warning') print(f"Password reset URL for {user.email}: {reset_url}") print(f"Email error: {str(e)}") # Log the activity log_activity(user.id, 'password_reset_request', f"Password reset requested for: {user.email}") return redirect(url_for('auth.login')) # Always show the same message whether user exists or not (security) flash('If your email is registered, you will receive a password reset link', 'info') return redirect(url_for('auth.login')) return render_template('auth/reset_password_request.html') # Route for resetting password with a token @auth_bp.route('/reset-password/', methods=['GET', 'POST']) def reset_password(token): # Find the reset token reset_token = PasswordReset.query.filter_by(token=token).first() if not reset_token or not reset_token.is_valid: flash('Invalid or expired password reset link', 'danger') return redirect(url_for('auth.login')) if request.method == 'POST': password = request.form.get('password') confirm_password = request.form.get('confirm_password') if not password or not confirm_password: flash('Please enter both fields', 'danger') return render_template('auth/reset_password.html', token=token) if password != confirm_password: flash('Passwords do not match', 'danger') return render_template('auth/reset_password.html', token=token) # Update the user's password user = reset_token.user user.set_password(password) # Mark the token as used reset_token.used = True db_session.commit() # Log the activity log_activity(user.id, 'password_reset', f"Password reset for user: {user.email}") flash('Your password has been reset. You can now log in with your new password.', 'success') return redirect(url_for('auth.login')) return render_template('auth/reset_password.html', token=token) # Create a context processor to make the current user available in templates @auth_bp.app_context_processor def inject_user(): if 'user_id' in session: user = User.query.get(session['user_id']) return {'current_user': user} return {'current_user': None} # Before each request, check session validity @auth_bp.before_app_request def check_session_validity(): if 'user_id' in session and 'session_token' in session: # Get the session from the database user_session = Session.query.filter_by( user_id=session['user_id'], session_token=session['session_token'] ).first() # If session is expired or doesn't exist, log the user out if not user_session or user_session.is_expired: session.clear()