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

"""Authentication API endpoints."""

import time
from datetime import datetime, timedelta, timezone

from flask import Blueprint, request, current_app
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, generate_password_hash

from shopdb.extensions import db, cache
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


def _login_ip():
    """Caller IP for rate limiting, honoring the first X-Forwarded-For hop."""
    forwarded = request.headers.get('X-Forwarded-For')
    if forwarded:
        return forwarded.split(',')[0].strip()
    return request.remote_addr or 'unknown'


def _login_ratelimited():
    """Fixed-window per-IP login limiter. Returns True when the caller is over
    budget for the current window.

    Backed by the existing cache extension (no new dependency). Under the
    default SimpleCache the counter is per-process, so with N gunicorn workers
    the effective budget is N x AUTH_RATELIMIT_MAX. This is defense in depth
    layered on top of the per-account lockout (see login()); a shared cache
    backend (Redis/memcached) tightens it to a true global budget.
    """
    if not current_app.config.get('AUTH_RATELIMIT_ENABLED', True):
        return False
    window = current_app.config.get('AUTH_RATELIMIT_WINDOW_SECONDS', 300)
    maxhits = current_app.config.get('AUTH_RATELIMIT_MAX', 30)
    # Time bucket makes this a fixed window: the key rolls over at each window
    # boundary, so a per-hit set() cannot turn it into a sliding window.
    bucket = int(time.time() // window) if window > 0 else 0
    key = f'loginratelimit:{_login_ip()}:{bucket}'
    count = cache.get(key) or 0
    if count >= maxhits:
        return True
    cache.set(key, count + 1, timeout=window)
    return False


@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": {...}
            }
        }
    """
    if _login_ratelimited():
        return error_response(
            'RATE_LIMITED',
            'Too many login attempts. Try again later.',
            http_code=429
        )

    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:
                # Naive UTC to match the naive lockeduntil column comparisons.
                user.lockeduntil = datetime.now(timezone.utc).replace(tzinfo=None) \
                    + 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],
            'permissions': user.getpermissions(),
            'mustchangepassword': bool(user.mustchangepassword)
        }
    })


@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
    """Refresh access token using refresh token."""
    user_id = get_jwt_identity()
    user = db.session.get(User, 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(),
        'mustchangepassword': bool(current_user.mustchangepassword)
    })


@auth_bp.route('/change-password', methods=['POST'])
@jwt_required()
def change_password():
    """Change the authenticated user's own password.

    Request:
        { "current_password": "string", "new_password": "string" }

    current_password is required for a normal self-service change. When the
    account is flagged mustchangepassword (an admin set a temporary password),
    the forced-change case accepts new_password alone. On success the flag is
    cleared and any lockout/failed-login state is reset.
    """
    data = request.get_json() or {}
    new_password = data.get('new_password')
    current_password = data.get('current_password')

    if not new_password:
        return error_response(
            ErrorCodes.VALIDATION_ERROR, 'new_password is required')
    if len(new_password) < 8:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'New password must be at least 8 characters')

    user = current_user

    # A normal change must prove knowledge of the current password. The forced
    # first-login case (admin-set temp password) may skip it.
    if not user.mustchangepassword:
        if not current_password:
            return error_response(
                ErrorCodes.VALIDATION_ERROR, 'current_password is required')
        if not check_password_hash(user.passwordhash, current_password):
            return error_response(
                ErrorCodes.UNAUTHORIZED,
                'Current password is incorrect', http_code=401)

    user.passwordhash = generate_password_hash(new_password)
    user.mustchangepassword = False
    user.failedlogins = 0
    user.lockeduntil = None
    db.session.commit()

    return success_response(
        {'mustchangepassword': False}, message='Password changed')


@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')