kanban-app/backend/app/services/auth_service.py

"""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()