""" User Management Blueprint Handles user CRUD operations """ from flask import Blueprint, render_template, request, jsonify, send_file, session from werkzeug.security import check_password_hash from blueprints.auth import login_required, can_user, get_current_user import data_access as da from io import BytesIO import json users_bp = Blueprint('users', __name__, url_prefix='/users') def permission_required(permission): """Decorator to require specific permission""" from functools import wraps def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): if not can_user(permission): return jsonify({ 'status': 'error', 'message': 'Permission denied' }), 403 return f(*args, **kwargs) return decorated_function return decorator # Routes @users_bp.route('/') @login_required def user_manager(): """Serve the user management page""" if not can_user('manage_users'): return "

403 - Permission Denied

You don't have permission to manage users.

", 403 return render_template('user_manager.html') @users_bp.route('/api', methods=['GET']) @login_required @permission_required('manage_users') def get_users(): """Get all users""" users = da.get_all_users() return jsonify({ 'status': 'success', 'users': users }) @users_bp.route('/api', methods=['POST']) @login_required @permission_required('manage_users') def add_user(): """Add a new user with hashed password""" try: data = request.json # Validate required fields if not data.get('username') or not data.get('password') or not data.get('defaultLocation'): return jsonify({ 'status': 'error', 'message': 'Username, password, and default location are required' }), 400 # Create user via data access layer new_user = da.create_user(data) if not new_user: return jsonify({ 'status': 'error', 'message': 'Username already exists' }), 400 users = da.get_all_users() return jsonify({ 'status': 'success', 'message': 'User added successfully', 'users': users }) except Exception as e: da.rollback() return jsonify({ 'status': 'error', 'message': str(e) }), 500 @users_bp.route('/api/', methods=['DELETE']) @login_required @permission_required('manage_users') def delete_user(user_id): """Delete a specific user by ID""" try: if not da.delete_user(user_id): return jsonify({ 'status': 'error', 'message': 'User not found' }), 404 users = da.get_all_users() return jsonify({ 'status': 'success', 'message': 'User deleted successfully', 'users': users }) except Exception as e: da.rollback() return jsonify({ 'status': 'error', 'message': str(e) }), 500 @users_bp.route('/api//active', methods=['PATCH']) @login_required @permission_required('manage_users') def toggle_user_active(user_id): """Toggle user active status""" try: data = request.json user = da.toggle_user_active(user_id, data.get('active', True)) if not user: return jsonify({ 'status': 'error', 'message': 'User not found' }), 404 users = da.get_all_users() return jsonify({ 'status': 'success', 'message': 'User status updated', 'users': users }) except Exception as e: da.rollback() return jsonify({ 'status': 'error', 'message': str(e) }), 500 @users_bp.route('/api//change-password', methods=['POST']) @login_required @permission_required('manage_users') def change_user_password(user_id): """Change a user's password with current user verification""" try: data = request.json new_password = data.get('newPassword') current_user_password = data.get('currentUserPassword') if not new_password or not current_user_password: return jsonify({ 'status': 'error', 'message': 'New password and current user password are required' }), 400 if len(new_password) < 6: return jsonify({ 'status': 'error', 'message': 'Password must be at least 6 characters long' }), 400 # Verify current user's password current_user = get_current_user() if not current_user: return jsonify({ 'status': 'error', 'message': 'Not authenticated' }), 401 if not check_password_hash(current_user['password'], current_user_password): return jsonify({ 'status': 'error', 'message': 'Your password is incorrect. Password change denied for security reasons.' }), 403 # Update password via data access layer updated_user = da.change_user_password(user_id, new_password) if not updated_user: return jsonify({ 'status': 'error', 'message': 'User not found' }), 404 return jsonify({ 'status': 'success', 'message': f'Password changed successfully for user "{updated_user["username"]}"' }) except Exception as e: da.rollback() return jsonify({ 'status': 'error', 'message': str(e) }), 500 @users_bp.route('/api/download') @login_required @permission_required('manage_users') def download_users(): """Download users as JSON file""" try: users = da.get_all_users() if not users: return jsonify({ 'status': 'error', 'message': 'No users to download' }), 400 json_data = json.dumps(users, indent=2) return send_file( BytesIO(json_data.encode()), mimetype='application/json', as_attachment=True, download_name='users.json' ) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500