import logging from datetime import datetime from urllib.parse import urlparse, urljoin from flask import Blueprint, render_template, redirect, url_for, flash, request from flask_login import login_user, logout_user, login_required, current_user from app import db, limiter from app.models import User, UserRole from app.services.log_service import log_action auth_bp = Blueprint('auth', __name__, url_prefix='/auth') logger = logging.getLogger(__name__) def _is_safe_url(target): """Return True only when *target* points back to this same host. Prevents open-redirect attacks where an attacker crafts a login URL with ?next=https://evil.com — without this check the user would be silently forwarded to an external site after authentication. """ 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 ) @auth_bp.route('/login', methods=['GET', 'POST']) @limiter.limit('10 per minute; 50 per hour') def login(): if current_user.is_authenticated: return redirect(url_for('tickets.dashboard')) if request.method == 'POST': email = request.form.get('email', '').strip().lower() password = request.form.get('password', '') remember = bool(request.form.get('remember')) user = User.query.filter_by(email=email).first() if user and user.check_password(password) and user.is_active: login_user(user, remember=remember) user.last_login = datetime.utcnow() log_action(user.id, 'user_login', 'user', user.id, f'email={email}') db.session.commit() logger.info(f'[AUTH LOGIN] user_id={user.id} email={email}') next_page = request.args.get('next') if next_page and not _is_safe_url(next_page): logger.warning(f'[AUTH OPEN-REDIRECT BLOCKED] next={next_page} user_id={user.id}') next_page = None return redirect(next_page or url_for('tickets.dashboard')) else: logger.warning(f'[AUTH FAILED] email={email} ip={request.remote_addr}') flash('Invalid credentials or account disabled.', 'danger') return render_template('auth/login.html') @auth_bp.route('/register', methods=['GET', 'POST']) @limiter.limit('5 per minute; 20 per hour') def register(): if current_user.is_authenticated: return redirect(url_for('tickets.dashboard')) if request.method == 'POST': email = request.form.get('email', '').strip().lower() username = request.form.get('username', '').strip() full_name = request.form.get('full_name', '').strip() department = request.form.get('department', '').strip() phone = request.form.get('phone', '').strip() password = request.form.get('password', '') confirm = request.form.get('confirm_password', '') if User.query.filter_by(email=email).first(): flash('Email already registered.', 'danger') elif User.query.filter_by(username=username).first(): flash('Username already taken.', 'danger') elif password != confirm: flash('Passwords do not match.', 'danger') elif len(password) < 8: flash('Password must be at least 8 characters.', 'danger') else: user = User( email = email, username = username, full_name = full_name, department = department, phone = phone, role = UserRole.EMPLOYEE, ) user.set_password(password) db.session.add(user) log_action(user.id, 'user_register', 'user', user.id, f'email={email}') db.session.commit() logger.info(f'[AUTH REGISTER] user_id={user.id} email={email}') flash('Account created! You may now log in.', 'success') return redirect(url_for('auth.login')) return render_template('auth/register.html') @auth_bp.route('/logout') @login_required def logout(): log_action(current_user.id, 'user_logout', 'user', current_user.id) logger.info(f'[AUTH LOGOUT] user_id={current_user.id}') logout_user() flash('You have been logged out.', 'info') return redirect(url_for('auth.login')) @auth_bp.route('/profile', methods=['GET', 'POST']) @login_required def profile(): if request.method == 'POST': full_name = request.form.get('full_name', '').strip() department = request.form.get('department', '').strip() phone = request.form.get('phone', '').strip() email_notif= bool(request.form.get('email_notif')) web_notif = bool(request.form.get('web_notif')) new_pw = request.form.get('new_password', '') confirm_pw = request.form.get('confirm_password', '') current_user.full_name = full_name current_user.department = department current_user.phone = phone current_user.email_notif= email_notif current_user.web_notif = web_notif if new_pw: if new_pw != confirm_pw: flash('Passwords do not match.', 'danger') return render_template('auth/profile.html') if len(new_pw) < 8: flash('Password must be at least 8 characters.', 'danger') return render_template('auth/profile.html') current_user.set_password(new_pw) logger.info(f'[AUTH PASSWORD CHANGE] user_id={current_user.id}') log_action(current_user.id, 'user_profile_update', 'user', current_user.id) db.session.commit() logger.info(f'[AUTH PROFILE UPDATE] user_id={current_user.id}') flash('Profile updated successfully.', 'success') return render_template('auth/profile.html')