import logging import os import uuid from datetime import datetime from urllib.parse import urlparse, urljoin from flask import Blueprint, render_template, redirect, url_for, flash, request, current_app, send_from_directory from flask_login import login_user, logout_user, login_required, current_user from werkzeug.utils import secure_filename from app import db, limiter from app.models import User, UserRole from app.services.log_service import log_action from app.services.validation_service import validate_password, validate_file auth_bp = Blueprint('auth', __name__, url_prefix='/auth') logger = logging.getLogger(__name__) AVATAR_ALLOWED_EXT = {'png', 'jpg', 'jpeg', 'gif', 'webp'} AVATAR_MAX_BYTES = 5 * 1024 * 1024 # 5 MB LOGO_ALLOWED_EXT = {'png', 'jpg', 'jpeg', 'gif', 'webp', 'svg'} LOGO_MAX_BYTES = 2 * 1024 * 1024 # 2 MB 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')) from app.models import SystemSetting if not SystemSetting.get_bool('registration_enabled', default=True): logger.info(f'[AUTH REGISTER BLOCKED] Registration is disabled. ip={request.remote_addr}') flash('Self-registration is currently disabled. Please contact your IT administrator.', 'warning') return redirect(url_for('auth.login')) 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') else: pw_error = validate_password(password, confirm) if pw_error: flash(pw_error, '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) # Flush to obtain user.id from the DB sequence before logging. # Without this flush, user.id is None and the activity log entry # records entity_id=None, making the log entry unlinkable. db.session.flush() 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 # ── Avatar upload ───────────────────────────────────────────────────── avatar_file = request.files.get('avatar') if avatar_file and avatar_file.filename: file_error = validate_file(avatar_file, AVATAR_ALLOWED_EXT) if file_error: flash(f'Avatar not saved: {file_error}', 'danger') else: avatar_file.stream.seek(0, 2) avatar_size = avatar_file.stream.tell() avatar_file.stream.seek(0) if avatar_size > AVATAR_MAX_BYTES: flash('Avatar image must be under 5 MB.', 'danger') else: ext = secure_filename(avatar_file.filename).rsplit('.', 1)[-1].lower() stored_name = f"avatar_{current_user.id}_{uuid.uuid4().hex}.{ext}" upload_dir = current_app.config['UPLOAD_FOLDER'] # Delete old avatar file from disk if present if current_user.avatar_url: old_file = os.path.join(upload_dir, os.path.basename(current_user.avatar_url)) if os.path.exists(old_file): os.remove(old_file) avatar_file.save(os.path.join(upload_dir, stored_name)) current_user.avatar_url = stored_name logger.info(f'[AUTH AVATAR UPLOAD] user_id={current_user.id} file={stored_name}') if new_pw: pw_error = validate_password(new_pw, confirm_pw) if pw_error: flash(pw_error, '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') @auth_bp.route('/avatar/') @login_required def serve_avatar(filename): """Serve a user avatar image stored in the upload folder.""" # Prevent path traversal — stored_name never contains slashes if '/' in filename or '\\' in filename or '..' in filename: from flask import abort abort(400) upload_dir = current_app.config['UPLOAD_FOLDER'] return send_from_directory(upload_dir, filename, as_attachment=False) @auth_bp.route('/logo/') def serve_logo(filename): """Serve the company logo — publicly accessible (shown on login page).""" if '/' in filename or '\\' in filename or '..' in filename: abort(400) upload_dir = current_app.config['UPLOAD_FOLDER'] return send_from_directory(upload_dir, filename, as_attachment=False)