331 lines
14 KiB
Python
331 lines
14 KiB
Python
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, current_user.password_hash)
|
|
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('/forgot-password', methods=['GET', 'POST'])
|
|
@limiter.limit('5 per hour')
|
|
def forgot_password():
|
|
"""Show the forgot-password form and send a reset email on POST."""
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('tickets.dashboard'))
|
|
|
|
if request.method == 'POST':
|
|
email = request.form.get('email', '').strip().lower()
|
|
user = User.query.filter_by(email=email, is_active=True).first()
|
|
|
|
# Always show the same message — prevents user enumeration
|
|
flash('If that email is registered, a password reset link has been sent.', 'info')
|
|
|
|
if user:
|
|
from app.models import PasswordResetToken
|
|
# Invalidate any existing unused tokens for this user
|
|
PasswordResetToken.query.filter_by(user_id=user.id).delete()
|
|
raw, _token = PasswordResetToken.generate(user.id)
|
|
db.session.commit()
|
|
logger.info(f'[AUTH RESET REQUEST] user_id={user.id} email={email}')
|
|
|
|
reset_url = url_for('auth.reset_password', token=raw, _external=True)
|
|
_send_reset_email(user, reset_url)
|
|
|
|
return redirect(url_for('auth.login'))
|
|
|
|
return render_template('auth/forgot_password.html')
|
|
|
|
|
|
@auth_bp.route('/reset-password/<token>', methods=['GET', 'POST'])
|
|
@limiter.limit('10 per hour')
|
|
def reset_password(token):
|
|
"""Validate the reset token and allow the user to set a new password."""
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('tickets.dashboard'))
|
|
|
|
from app.models import PasswordResetToken
|
|
token_row = PasswordResetToken.verify(token)
|
|
if not token_row:
|
|
flash('This password reset link is invalid or has expired. Please request a new one.', 'danger')
|
|
return redirect(url_for('auth.forgot_password'))
|
|
|
|
if request.method == 'POST':
|
|
password = request.form.get('password', '')
|
|
confirm = request.form.get('confirm_password', '')
|
|
user = token_row.user
|
|
pw_error = validate_password(password, confirm, user.password_hash)
|
|
if pw_error:
|
|
flash(pw_error, 'danger')
|
|
return render_template('auth/reset_password.html', token=token)
|
|
|
|
user.set_password(password)
|
|
db.session.delete(token_row) # single-use — delete immediately
|
|
log_action(user.id, 'password_reset', 'user', user.id)
|
|
db.session.commit()
|
|
logger.info(f'[AUTH RESET COMPLETE] user_id={user.id}')
|
|
flash('Password updated successfully. You may now log in.', 'success')
|
|
return redirect(url_for('auth.login'))
|
|
|
|
return render_template('auth/reset_password.html', token=token)
|
|
|
|
|
|
def _send_reset_email(user, reset_url):
|
|
"""Send the password reset email in a background thread."""
|
|
from threading import Thread
|
|
from flask_mail import Message
|
|
from app import mail
|
|
|
|
html = f"""
|
|
<html><body style="font-family:Arial,sans-serif;background:#f4f4f4;padding:20px;">
|
|
<div style="max-width:520px;margin:0 auto;background:#fff;border-radius:10px;
|
|
overflow:hidden;box-shadow:0 2px 12px rgba(0,0,0,.08);">
|
|
<div style="background:#1e293b;padding:24px 32px;">
|
|
<h1 style="color:#fff;margin:0;font-size:20px;">🔐 Password Reset Request</h1>
|
|
</div>
|
|
<div style="padding:32px;">
|
|
<p style="color:#334155;margin-top:0;">Hi {user.full_name},</p>
|
|
<p style="color:#334155;">We received a request to reset your TechDesk password.
|
|
Click the button below to choose a new one.</p>
|
|
<p style="text-align:center;margin:28px 0;">
|
|
<a href="{reset_url}"
|
|
style="display:inline-block;background:#2563eb;color:#fff;padding:13px 32px;
|
|
border-radius:8px;text-decoration:none;font-weight:600;font-size:15px;">
|
|
Reset My Password
|
|
</a>
|
|
</p>
|
|
<p style="color:#64748b;font-size:13px;">This link expires in <strong>1 hour</strong>.
|
|
If you did not request a password reset, you can safely ignore this email.</p>
|
|
<p style="color:#64748b;font-size:12px;word-break:break-all;">
|
|
Or paste this link into your browser:<br/>{reset_url}
|
|
</p>
|
|
</div>
|
|
<div style="background:#f8fafc;padding:16px 32px;text-align:center;
|
|
color:#94a3b8;font-size:11px;border-top:1px solid #e2e8f0;">
|
|
TechDesk IT Helpdesk • This is an automated message.
|
|
</div>
|
|
</div>
|
|
</body></html>"""
|
|
|
|
msg = Message(
|
|
subject = 'TechDesk — Password Reset Request',
|
|
recipients = [user.email],
|
|
html = html,
|
|
)
|
|
|
|
def _send():
|
|
from flask import current_app
|
|
with current_app.app_context():
|
|
try:
|
|
mail.send(msg)
|
|
logger.info(f'[AUTH RESET EMAIL SENT] user_id={user.id}')
|
|
except Exception as exc:
|
|
logger.error(f'[AUTH RESET EMAIL FAILED] user_id={user.id} {exc}')
|
|
|
|
Thread(target=_send, daemon=True).start()
|
|
|
|
|
|
@auth_bp.route('/avatar/<string:filename>')
|
|
@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/<string:filename>')
|
|
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)
|
|
|