Files

234 lines
9.2 KiB
Python

"""
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, register_user,
)
from config import get_setting
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."
reg_enabled = get_setting("registration.enabled", "0") == "1"
return render_template("login.html", error=error, reg_enabled=reg_enabled)
@auth_bp.route("/register", methods=["GET", "POST"])
def register():
if "user" in session:
return redirect(url_for("index"))
if get_setting("registration.enabled", "0") != "1":
flash("Registration is currently disabled. Contact your administrator.", "warning")
return redirect(url_for("auth.login"))
if request.method == "GET":
return render_template("register.html")
username = request.form.get("username", "").strip()
full_name = request.form.get("full_name", "").strip()
email = request.form.get("email", "").strip().lower()
password = request.form.get("password", "")
confirm = request.form.get("confirm_password", "")
errors = []
if not username or len(username) < 3:
errors.append("Username must be at least 3 characters.")
if len(password) < 8:
errors.append("Password must be at least 8 characters.")
if password != confirm:
errors.append("Passwords do not match.")
if errors:
return render_template("register.html", errors=errors,
username=username, full_name=full_name, email=email)
try:
new_id = register_user(username, password, full_name, email or None)
logger.info(f"New user registered: '{username}' id={new_id}.")
flash("Account created successfully. You can now sign in.", "success")
return redirect(url_for("auth.login"))
except ValueError as e:
return render_template("register.html", errors=[str(e)],
username=username, full_name=full_name, email=email)
except Exception as e:
logger.error(f"register error: {e}")
return render_template("register.html",
errors=["An unexpected error occurred. Please try again."],
username=username, full_name=full_name, email=email)
@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 Bid 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"], "Bid 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/<token>", 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"))