from functools import wraps from flask import jsonify, request, current_app from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request from app.models import User, Permission from .extensions import db class Authorization: @staticmethod def check_permission(required_permission): def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): from .models import User # 在函数内部导入以避免循环导入 current_user_id = get_jwt_identity() user = User.query.get(current_user_id) if not user: return jsonify({"message": "User not found"}), 404 user_permissions = user.get_all_permissions() if required_permission not in user_permissions: return jsonify({"message": "Permission denied"}), 403 return f(*args, **kwargs) return decorated_function return decorator @staticmethod def check_role(required_role): def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): current_user_id = get_jwt_identity() user = User.query.get(current_user_id) if not user: return jsonify({"message": "User not found"}), 404 user_roles = [role.name for role in user.roles] if required_role not in user_roles: return jsonify({"message": "Role not authorized"}), 403 return f(*args, **kwargs) return decorated_function return decorator @staticmethod def check_department(required_department): def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): current_user_id = get_jwt_identity() user = User.query.get(current_user_id) if not user: return jsonify({"message": "User not found"}), 404 user_departments = [user.primary_department.name] + [dept.name for dept in user.secondary_departments] if required_department not in user_departments: return jsonify({"message": "Department not authorized"}), 403 return f(*args, **kwargs) return decorated_function return decorator @staticmethod def api_permission_required(permission): def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): verify_jwt_in_request() current_user_id = get_jwt_identity() user = User.query.get(current_user_id) if not user: return jsonify({"message": "User not found"}), 404 user_permissions = user.get_all_permissions() if permission not in user_permissions: return jsonify({"message": "Permission denied"}), 403 return f(*args, **kwargs) return decorated_function return decorator def init_permissions(): from .models import Permission # 在函数内部导入以避免循环导入 # 这个函数可以在应用启动时调用,用于初始化或更新权限 # 可以从配置文件、数据库或其他来源读取权限定义 permissions = [ "create_user", "edit_user", "delete_user", "create_role", "edit_role", "delete_role", "create_department", "edit_department", "delete_department", # ... 其他权限 ... ] for perm_name in permissions: perm = Permission.query.filter_by(name=perm_name).first() if not perm: new_perm = Permission(name=perm_name) db.session.add(new_perm) db.session.commit() # 可以添加更多的辅助函数,比如检查复杂的权限组合等