208 lines
6.3 KiB
Python
208 lines
6.3 KiB
Python
"""Authentication service for user registration, login, password management, and lockout"""
|
|
|
|
from datetime import UTC, datetime, timedelta, timezone
|
|
|
|
from flask import current_app
|
|
from flask_jwt_extended import create_access_token, create_refresh_token
|
|
|
|
from app import db
|
|
from app.models import LoginHistory, User
|
|
|
|
|
|
def register_user(data):
|
|
"""Register a new user with server-side password confirmation validation.
|
|
|
|
Args:
|
|
data: dict with email, password, confirm_password, username, etc.
|
|
|
|
Returns:
|
|
User object on success
|
|
|
|
Raises:
|
|
ValueError: If validation fails
|
|
"""
|
|
email = data.get("email", "").strip().lower()
|
|
password = data.get("password", "")
|
|
confirm_password = data.get("confirm_password", "")
|
|
username = data.get("username", email.split("@")[0] if email else "")
|
|
|
|
if not email:
|
|
raise ValueError("Email is required")
|
|
|
|
if not password:
|
|
raise ValueError("Password is required")
|
|
|
|
if password != confirm_password:
|
|
raise ValueError("Passwords do not match")
|
|
|
|
if len(password) < 6:
|
|
raise ValueError("Password must be at least 6 characters")
|
|
|
|
existing = User.query.filter_by(email=email).first()
|
|
if existing:
|
|
raise ValueError("Email already exists")
|
|
|
|
user = User(
|
|
email=email,
|
|
username=username,
|
|
first_name=data.get("first_name"),
|
|
last_name=data.get("last_name"),
|
|
)
|
|
user.set_password(password)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
|
|
return user
|
|
|
|
|
|
def authenticate_user(email, password, ip_address=None, user_agent=None):
|
|
"""Authenticate user with account lockout logic.
|
|
|
|
Implements two lockout strategies:
|
|
- "timeout": locks account for ACCOUNT_LOCK_DURATION_MINUTES
|
|
- "permanent": sets is_active to False (requires admin intervention)
|
|
|
|
Args:
|
|
email: User email
|
|
password: Plain text password
|
|
ip_address: Request IP address (for logging)
|
|
user_agent: Request user agent (for logging)
|
|
|
|
Returns:
|
|
(user, access_token, refresh_token) on success
|
|
|
|
Raises:
|
|
ValueError: With appropriate error message on failure
|
|
"""
|
|
max_attempts = current_app.config["MAX_FAILED_LOGIN_ATTEMPTS"]
|
|
lock_duration = current_app.config["ACCOUNT_LOCK_DURATION_MINUTES"]
|
|
lockout_strategy = current_app.config["LOCKOUT_STRATEGY"]
|
|
|
|
user = User.query.filter_by(email=email.strip().lower()).first()
|
|
|
|
if not user:
|
|
raise ValueError("Invalid email or password")
|
|
|
|
# Check permanent lock (is_active=False)
|
|
if not user.is_active:
|
|
_record_login_history(user, ip_address, user_agent, success=False)
|
|
raise ValueError(
|
|
"Account has been locked. Contact an administrator."
|
|
)
|
|
|
|
# Check timeout lock
|
|
if user.locked_until and user.locked_until > datetime.now(timezone.utc):
|
|
remaining_minutes = int(
|
|
(user.locked_until - datetime.now(timezone.utc)).total_seconds() / 60
|
|
)
|
|
_record_login_history(user, ip_address, user_agent, success=False)
|
|
raise ValueError(
|
|
f"Account is locked. Try again in {remaining_minutes} minute(s)."
|
|
)
|
|
|
|
if user.check_password(password):
|
|
# Success: reset lockout counters
|
|
user.failed_login_attempts = 0
|
|
user.locked_until = None
|
|
user.last_login_at = datetime.now(timezone.utc)
|
|
db.session.commit()
|
|
|
|
_record_login_history(user, ip_address, user_agent, success=True)
|
|
|
|
access_token = create_access_token(identity=str(user.id))
|
|
refresh_token = create_refresh_token(identity=str(user.id))
|
|
|
|
return user, access_token, refresh_token
|
|
|
|
# Failed login
|
|
user.failed_login_attempts = (user.failed_login_attempts or 0) + 1
|
|
_record_login_history(user, ip_address, user_agent, success=False)
|
|
|
|
if user.failed_login_attempts >= max_attempts:
|
|
if lockout_strategy == "permanent":
|
|
user.is_active = False
|
|
user.locked_until = None
|
|
db.session.commit()
|
|
raise ValueError(
|
|
"Too many failed attempts. Account has been permanently locked. "
|
|
"Contact an administrator."
|
|
)
|
|
else:
|
|
# Default: timeout
|
|
user.locked_until = datetime.now(timezone.utc) + timedelta(
|
|
minutes=lock_duration
|
|
)
|
|
db.session.commit()
|
|
raise ValueError(
|
|
f"Too many failed attempts. Account locked for "
|
|
f"{lock_duration} minute(s)."
|
|
)
|
|
|
|
db.session.commit()
|
|
raise ValueError("Invalid email or password")
|
|
|
|
|
|
def change_password(user, current_password, new_password, confirm_password):
|
|
"""Change a user's password after verifying current password.
|
|
|
|
Args:
|
|
user: User object
|
|
current_password: Current password for verification
|
|
new_password: New password
|
|
confirm_password: Confirmation of new password
|
|
|
|
Raises:
|
|
ValueError: If validation fails
|
|
"""
|
|
if not user.check_password(current_password):
|
|
raise ValueError("Current password is incorrect")
|
|
|
|
if new_password != confirm_password:
|
|
raise ValueError("New passwords do not match")
|
|
|
|
if len(new_password) < 6:
|
|
raise ValueError("New password must be at least 6 characters")
|
|
|
|
if current_password == new_password:
|
|
raise ValueError("New password must be different from current password")
|
|
|
|
user.set_password(new_password)
|
|
db.session.commit()
|
|
|
|
|
|
def get_login_history(user_id, limit=20):
|
|
"""Get recent login history for a user.
|
|
|
|
Args:
|
|
user_id: User ID
|
|
limit: Maximum number of entries to return
|
|
|
|
Returns:
|
|
List of LoginHistory dicts, most recent first
|
|
"""
|
|
entries = (
|
|
LoginHistory.query.filter_by(user_id=user_id)
|
|
.order_by(LoginHistory.created_at.desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return [entry.to_dict() for entry in entries]
|
|
|
|
|
|
def _record_login_history(user, ip_address, user_agent, success):
|
|
"""Record a login attempt in the login history.
|
|
|
|
Args:
|
|
user: User object
|
|
ip_address: Request IP address
|
|
user_agent: Request user agent
|
|
success: Whether the login was successful
|
|
"""
|
|
entry = LoginHistory(
|
|
user_id=user.id,
|
|
ip_address=ip_address or "unknown",
|
|
user_agent=user_agent or "unknown",
|
|
success=success,
|
|
)
|
|
db.session.add(entry)
|
|
db.session.commit()
|