""" routes/auth.py — Authentication routes: login, logout, change-password. """ import logging from flask import Blueprint, render_template, request, session, redirect, url_for, flash from models import ( authenticate, check_login_allowed, change_password, log_action, get_user_by_email, create_password_reset_token, get_password_reset_user, consume_password_reset_token, purge_expired_reset_tokens, update_user_profile, ) from utils.decorators import login_required from utils.email import send_email logger = logging.getLogger("routes.auth") auth_bp = Blueprint("auth", __name__) @auth_bp.route("/login", methods=["GET", "POST"]) def login(): if "user" in session: return redirect(url_for("index")) error = None if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "") ip = request.remote_addr allowed, seconds_remaining = check_login_allowed(username) if not allowed: mins = seconds_remaining // 60 secs = seconds_remaining % 60 error = f"Account locked. Try again in {mins}m {secs}s." logger.warning(f"Login blocked for '{username}' — still locked ({seconds_remaining}s remaining).") else: user = authenticate(username, password, ip_address=ip) if user: session.clear() # prevent session fixation session.permanent = True # Store a safe subset — never store the password hash in session session["user"] = { "id": user["id"], "username": user["username"], "full_name": user.get("full_name") or user["username"], "role": user["role"], "email": user.get("email", ""), } try: purge_expired_reset_tokens() except Exception: pass # non-critical cleanup; never fail a login because of it logger.info(f"User '{username}' logged in from {ip}.") flash(f"You have been signed in. Welcome, {session['user']['full_name']}!", "success") if user["role"] == "admin": return redirect(url_for("admin_dashboard.dashboard")) return redirect(url_for("user_dashboard.my_shifts")) else: error = "Invalid username or password." return render_template("login.html", error=error) @auth_bp.route("/logout") @login_required def logout(): user = session.get("user", {}) if user: log_action(user["id"], "LOGOUT", "users", user["id"], f"User '{user['username']}' logged out.") logger.info(f"User '{user['username']}' logged out.") session.clear() flash("You have been signed out.", "info") return redirect(url_for("auth.login")) @auth_bp.route("/change-password", methods=["GET", "POST"]) @login_required def change_password_view(): user = session["user"] success = None error = None if request.method == "POST": old_pw = request.form.get("old_password", "") new_pw = request.form.get("new_password", "") confirm = request.form.get("confirm_password", "") if new_pw != confirm: error = "New password and confirmation do not match." else: ok, msg = change_password(user["id"], old_pw, new_pw) if ok: success = msg else: error = msg return render_template("change_password.html", success=success, error=error) @auth_bp.route("/ping") @login_required def ping(): """Keep-alive endpoint for the session timeout warning in app.js.""" session.modified = True return "", 204 @auth_bp.route("/profile", methods=["GET", "POST"]) @login_required def profile(): user = session["user"] if request.method == "POST": full_name = request.form.get("full_name", "").strip() email = request.form.get("email", "").strip() try: update_user_profile(user["id"], full_name, email) session["user"]["full_name"] = full_name or user["username"] session["user"]["email"] = email session.modified = True log_action(user["id"], "UPDATE_PROFILE", "users", user["id"], "User updated their own profile.") flash("Profile updated successfully.", "success") except Exception as e: logger.error(f"Profile update error for user_id={user['id']}: {e}") flash(f"Error: {e}", "danger") return redirect(url_for("auth.profile")) return render_template("profile.html") @auth_bp.route("/forgot-password", methods=["GET", "POST"]) def forgot_password(): if "user" in session: return redirect(url_for("index")) if request.method == "GET": return render_template("forgot_password.html") email = request.form.get("email", "").strip().lower() if email: user = get_user_by_email(email) if user and user.get("email"): try: token = create_password_reset_token(user["id"]) reset_url = url_for("auth.reset_password", token=token, _external=True) body = ( f"Hi {user.get('full_name') or user['username']},\n\n" f"A password reset was requested for your Website Checker account.\n" f"Click the link below to set a new password (valid for 1 hour):\n\n" f"{reset_url}\n\n" f"If you did not request this, you can safely ignore this email.\n" ) send_email(user["email"], "Website Checker — Password Reset", body) except Exception as e: logger.error(f"Password reset email error for '{email}': {e}") # Always show the same message to prevent user enumeration flash("If that email is registered, a reset link has been sent.", "info") return render_template("forgot_password.html") @auth_bp.route("/reset-password/", methods=["GET", "POST"]) def reset_password(token): user = get_password_reset_user(token) if not user: flash("This reset link is invalid or has expired.", "danger") return redirect(url_for("auth.forgot_password")) if request.method == "GET": return render_template("reset_password.html", token=token) password = request.form.get("password", "") confirm = request.form.get("confirm_password", "") if len(password) < 8: flash("Password must be at least 8 characters.", "danger") return render_template("reset_password.html", token=token) if password != confirm: flash("Passwords do not match.", "danger") return render_template("reset_password.html", token=token) if consume_password_reset_token(token, password): logger.info(f"Password reset successfully for user id={user['id']}.") flash("Password reset successfully. Please log in.", "success") return redirect(url_for("auth.login")) flash("Reset link expired or already used. Please request a new one.", "danger") return redirect(url_for("auth.forgot_password"))