shopdb-flask/shopdb/core/api/auth.py

"""Authentication API endpoints."""

from datetime import datetime, timedelta

from flask import Blueprint, request
from flask_jwt_extended import (
    create_access_token,
    create_refresh_token,
    jwt_required,
    get_jwt_identity,
    current_user
)
from werkzeug.security import check_password_hash

from shopdb.extensions import db
from shopdb.core.models import User
from shopdb.utils.responses import success_response, error_response, ErrorCodes

auth_bp = Blueprint('auth', __name__)

# Account lockout policy: after MAX_FAILED_LOGINS consecutive bad passwords,
# lock the account for LOCKOUT_MINUTES. Mitigates password brute-forcing.
MAX_FAILED_LOGINS = 5
LOCKOUT_MINUTES = 15


@auth_bp.route('/login', methods=['POST'])
def login():
    """
    Authenticate user and return JWT tokens.

    Request:
        {
            "username": "string",
            "password": "string"
        }

    Response:
        {
            "data": {
                "access_token": "...",
                "refresh_token": "...",
                "user": {...}
            }
        }
    """
    data = request.get_json()

    if not data or not data.get('username') or not data.get('password'):
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'Username and password required'
        )

    user = User.query.filter_by(
        username=data['username'],
        isactive=True
    ).first()

    # Reject a locked account before checking the password, so a lockout can't
    # be probed and a valid password can't reset the clock mid-lockout.
    if user and user.islocked:
        return error_response(
            ErrorCodes.FORBIDDEN,
            'Account is locked. Try again later or contact an administrator.',
            http_code=403
        )

    if not user or not check_password_hash(user.passwordhash, data['password']):
        # Count the failure and lock the account once the threshold is hit.
        # Only possible when the username matched a real account.
        if user:
            user.failedlogins = (user.failedlogins or 0) + 1
            if user.failedlogins >= MAX_FAILED_LOGINS:
                user.lockeduntil = datetime.utcnow() + timedelta(minutes=LOCKOUT_MINUTES)
                user.failedlogins = 0
            db.session.commit()
        return error_response(
            ErrorCodes.UNAUTHORIZED,
            'Invalid username or password',
            http_code=401
        )

    # Create tokens (identity must be a string in Flask-JWT-Extended 4.x)
    access_token = create_access_token(
        identity=str(user.userid),
        additional_claims={
            'username': user.username,
            'roles': [r.rolename for r in user.roles]
        }
    )
    refresh_token = create_refresh_token(identity=str(user.userid))

    # Update last login and clear any failed-login state
    user.lastlogindate = db.func.now()
    user.failedlogins = 0
    user.lockeduntil = None
    db.session.commit()

    return success_response({
        'access_token': access_token,
        'refresh_token': refresh_token,
        'token_type': 'Bearer',
        'expires_in': 3600,
        'user': {
            'userid': user.userid,
            'username': user.username,
            'email': user.email,
            'firstname': user.firstname,
            'lastname': user.lastname,
            'roles': [r.rolename for r in user.roles]
        }
    })


@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
    """Refresh access token using refresh token."""
    user_id = get_jwt_identity()
    user = User.query.get(int(user_id))

    if not user or not user.isactive:
        return error_response(
            ErrorCodes.UNAUTHORIZED,
            'User not found or inactive',
            http_code=401
        )

    access_token = create_access_token(
        identity=str(user.userid),
        additional_claims={
            'username': user.username,
            'roles': [r.rolename for r in user.roles]
        }
    )

    return success_response({
        'access_token': access_token,
        'token_type': 'Bearer',
        'expires_in': 3600
    })


@auth_bp.route('/me', methods=['GET'])
@jwt_required()
def get_current_user():
    """Get current authenticated user info."""
    return success_response({
        'userid': current_user.userid,
        'username': current_user.username,
        'email': current_user.email,
        'firstname': current_user.firstname,
        'lastname': current_user.lastname,
        'roles': [r.rolename for r in current_user.roles],
        'permissions': current_user.getpermissions()
    })


@auth_bp.route('/logout', methods=['POST'])
@jwt_required()
def logout():
    """Logout user (for frontend token cleanup)."""
    # In a full implementation, you'd blacklist the token
    return success_response(message='Successfully logged out')