"""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, port stripped so the key is per-host, not per-connection (ARR forwards clientip:port with an ephemeral port).""" from shopdb.utils.clientip import client_ip return client_ip(request) 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')