199 lines
7.0 KiB
Python
199 lines
7.0 KiB
Python
"""
|
|
app/tenant/auth/routes.py — Tenant portal authentication (email + password).
|
|
Routes: /login, /logout, /password-reset, /suspended, /cancelled
|
|
Roles: tenant_admin, tenant_manager
|
|
"""
|
|
|
|
import logging
|
|
import secrets
|
|
from datetime import datetime, timezone, timedelta
|
|
from flask import (
|
|
Blueprint, render_template, redirect, url_for,
|
|
flash, request, current_app, session, g,
|
|
)
|
|
from flask_login import login_user, logout_user, login_required, current_user
|
|
import bcrypt
|
|
|
|
from app.extensions import db, limiter, mail
|
|
from app.models.platform import Tenant
|
|
from app.models.salon import User
|
|
from flask_mail import Message
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
tenant_auth_bp = Blueprint("tenant_auth", __name__)
|
|
|
|
|
|
@tenant_auth_bp.route("/login", methods=["GET", "POST"])
|
|
@limiter.limit("10 per minute")
|
|
def login():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for("dashboard.index"))
|
|
|
|
error = None
|
|
|
|
if request.method == "POST":
|
|
email = request.form.get("email", "").strip().lower()
|
|
password = request.form.get("password", "")
|
|
|
|
user = User.query.filter_by(email=email, is_active=True)\
|
|
.filter(User.deleted_at.is_(None)).first()
|
|
|
|
max_attempts = current_app.config.get("MAX_LOGIN_ATTEMPTS", 5)
|
|
lockout_minutes = current_app.config.get("LOGIN_LOCKOUT_MINUTES", 15)
|
|
|
|
if user and user.is_locked():
|
|
logger.warning("Tenant login blocked — account locked: %s", email)
|
|
error = f"Account locked. Try again in {lockout_minutes} minutes."
|
|
|
|
elif user and bcrypt.checkpw(password.encode(), user.password_hash.encode()):
|
|
# Verify tenant is accessible
|
|
tenant = Tenant.query.get(user.tenant_id)
|
|
if not tenant or tenant.status == "cancelled":
|
|
error = "This account is no longer active."
|
|
else:
|
|
user.failed_login_attempts = 0
|
|
user.locked_until = None
|
|
user.last_login_at = datetime.now(timezone.utc)
|
|
db.session.commit()
|
|
|
|
login_user(user, remember=False)
|
|
session.permanent = True
|
|
|
|
logger.info(
|
|
"Tenant login success: user=%s tenant=%s",
|
|
user.id, user.tenant_id,
|
|
)
|
|
return redirect(url_for("dashboard.index"))
|
|
else:
|
|
if user:
|
|
user.failed_login_attempts += 1
|
|
if user.failed_login_attempts >= max_attempts:
|
|
user.locked_until = datetime.now(timezone.utc) + timedelta(
|
|
minutes=lockout_minutes
|
|
)
|
|
logger.warning(
|
|
"Tenant account locked after %d failures: %s",
|
|
max_attempts, email,
|
|
)
|
|
db.session.commit()
|
|
|
|
logger.warning("Tenant login failed: %s", email)
|
|
error = "Invalid email or password."
|
|
|
|
return render_template("tenant/auth/login.html", error=error)
|
|
|
|
|
|
@tenant_auth_bp.route("/logout")
|
|
@login_required
|
|
def logout():
|
|
logout_user()
|
|
session.clear()
|
|
logger.info("Tenant logout")
|
|
return redirect(url_for("tenant_auth.login"))
|
|
|
|
|
|
@tenant_auth_bp.route("/password-reset", methods=["GET", "POST"])
|
|
@limiter.limit("5 per hour")
|
|
def password_reset_request():
|
|
"""Step 1: submit email → send reset link."""
|
|
message = None
|
|
|
|
if request.method == "POST":
|
|
email = request.form.get("email", "").strip().lower()
|
|
user = User.query.filter_by(email=email)\
|
|
.filter(User.deleted_at.is_(None)).first()
|
|
|
|
if user:
|
|
token = secrets.token_urlsafe(32)
|
|
user.password_reset_token = token
|
|
user.password_reset_expires_at = datetime.now(timezone.utc) + timedelta(hours=1)
|
|
db.session.commit()
|
|
|
|
reset_url = url_for(
|
|
"tenant_auth.password_reset_confirm",
|
|
token=token,
|
|
_external=True,
|
|
)
|
|
try:
|
|
msg = Message(
|
|
subject="Reset your password",
|
|
recipients=[email],
|
|
body=f"Click the link below to reset your password (valid 1 hour):\n\n{reset_url}",
|
|
)
|
|
mail.send(msg)
|
|
logger.info("Password reset email sent to %s", email)
|
|
except Exception as exc:
|
|
logger.error("Failed to send reset email to %s: %s", email, exc)
|
|
|
|
# Always show the same message to prevent email enumeration
|
|
message = "If that email is registered, a reset link has been sent."
|
|
|
|
return render_template(
|
|
"tenant/auth/password_reset_request.html", message=message
|
|
)
|
|
|
|
|
|
@tenant_auth_bp.route("/password-reset/<token>", methods=["GET", "POST"])
|
|
def password_reset_confirm(token):
|
|
"""Step 2: confirm token → set new password."""
|
|
user = User.query.filter_by(password_reset_token=token)\
|
|
.filter(User.deleted_at.is_(None)).first()
|
|
|
|
now = datetime.now(timezone.utc)
|
|
|
|
if not user or not user.password_reset_expires_at:
|
|
flash("Invalid or expired reset link.", "danger")
|
|
return redirect(url_for("tenant_auth.login"))
|
|
|
|
expires = user.password_reset_expires_at
|
|
if expires.tzinfo is None:
|
|
expires = expires.replace(tzinfo=timezone.utc)
|
|
|
|
if now > expires:
|
|
flash("This reset link has expired. Please request a new one.", "danger")
|
|
return redirect(url_for("tenant_auth.password_reset_request"))
|
|
|
|
error = None
|
|
|
|
if request.method == "POST":
|
|
new_password = request.form.get("password", "")
|
|
confirm = request.form.get("confirm_password", "")
|
|
|
|
if new_password != confirm:
|
|
error = "Passwords do not match."
|
|
elif len(new_password) < 10:
|
|
error = "Password must be at least 10 characters."
|
|
elif not any(c.isupper() for c in new_password):
|
|
error = "Password must contain at least one uppercase letter."
|
|
elif not any(c.islower() for c in new_password):
|
|
error = "Password must contain at least one lowercase letter."
|
|
elif not any(c.isdigit() for c in new_password):
|
|
error = "Password must contain at least one digit."
|
|
else:
|
|
hashed = bcrypt.hashpw(new_password.encode(), bcrypt.gensalt(rounds=12))
|
|
user.password_hash = hashed.decode()
|
|
user.password_reset_token = None
|
|
user.password_reset_expires_at = None
|
|
user.failed_login_attempts = 0
|
|
user.locked_until = None
|
|
db.session.commit()
|
|
|
|
logger.info("Password reset completed for user %s", user.id)
|
|
flash("Password updated. Please log in.", "success")
|
|
return redirect(url_for("tenant_auth.login"))
|
|
|
|
return render_template(
|
|
"tenant/auth/password_reset_confirm.html", error=error, token=token
|
|
)
|
|
|
|
|
|
@tenant_auth_bp.route("/suspended")
|
|
def suspended():
|
|
return render_template("tenant/auth/suspended.html"), 403
|
|
|
|
|
|
@tenant_auth_bp.route("/cancelled")
|
|
def cancelled():
|
|
return render_template("tenant/auth/cancelled.html"), 403
|