本文目录导读：
我来介绍几种Python防越权的封装方案：
装饰器方式封装
基础装饰器实现
from functools import wraps
from enum import Enum
import hashlib
import json
class PermissionLevel(Enum):
GUEST = 0
USER = 1
ADMIN = 2
SUPER_ADMIN = 3
class PermissionDeniedError(Exception):
pass
def require_permission(required_level: PermissionLevel):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 从上下文中获取当前用户权限
current_user = get_current_user()
if current_user.permission_level.value < required_level.value:
raise PermissionDeniedError(
f"权限不足，需要{required_level.name}级别，当前为{current_user.permission_level.name}"
)
return func(*args, **kwargs)
return wrapper
return decorator
# 使用示例
@require_permission(PermissionLevel.ADMIN)
def delete_user(user_id):
# 删除用户逻辑
pass
带资源ID的权限校验
def check_resource_ownership(resource_type: str):
"""
检查资源所有权
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
current_user = get_current_user()
resource_id = kwargs.get(f'{resource_type}_id')
if not resource_id:
raise ValueError(f"缺少{resource_type}_id参数")
# 超级管理员可以访问所有资源
if current_user.permission_level >= PermissionLevel.SUPER_ADMIN:
return func(*args, **kwargs)
# 检查资源所有权
if not check_ownership(current_user.id, resource_type, resource_id):
raise PermissionDeniedError("无权访问该资源")
return func(*args, **kwargs)
return wrapper
return decorator
# 使用示例
@check_resource_ownership('document')
def view_document(document_id):
pass
类方法权限封装
class PermissionManager:
"""权限管理基类"""
def __init__(self, user):
self.user = user
self.permissions = self._load_permissions()
def _load_permissions(self):
"""加载用户权限配置"""
# 从数据库或配置文件加载
return self._get_permissions_from_db()
def _get_permissions_from_db(self):
# 模拟从数据库获取权限
pass
def has_permission(self, permission_name: str) -> bool:
"""检查是否有特定权限"""
return permission_name in self.permissions
def can_access_resource(self, resource_type: str, resource_id: int) -> bool:
"""检查是否能访问特定资源"""
if self.user.is_super_admin:
return True
# 检查资源所有权
resource = self._get_resource(resource_type, resource_id)
return resource.owner_id == self.user.id
class BaseController:
"""控制器基类"""
def __init__(self, request):
self.request = request
self.user = self._get_authenticated_user()
self.permission_manager = PermissionManager(self.user)
def _get_authenticated_user(self):
"""获取认证用户"""
# 从session或token获取
pass
def require_permission(self, permission_name: str):
"""要求特定权限"""
if not self.permission_manager.has_permission(permission_name):
raise PermissionDeniedError(f"缺少{permission_name}权限")
def require_ownership(self, resource_type: str, resource_id: int):
"""要求资源所有权"""
if not self.permission_manager.can_access_resource(resource_type, resource_id):
raise PermissionDeniedError("无权访问该资源")
RBAC（基于角色的访问控制）封装
class RBACManager:
"""RBAC权限管理系统"""
def __init__(self):
self.role_permissions = {}
self.user_roles = {}
def add_role(self, role_name: str, permissions: list):
"""添加角色和权限"""
self.role_permissions[role_name] = set(permissions)
def assign_role(self, user_id: int, role_name: str):
"""为用户分配角色"""
if user_id not in self.user_roles:
self.user_roles[user_id] = set()
self.user_roles[user_id].add(role_name)
def has_permission(self, user_id: int, permission: str) -> bool:
"""检查用户是否有特定权限"""
if user_id not in self.user_roles:
return False
for role in self.user_roles[user_id]:
if role in self.role_permissions:
if permission in self.role_permissions[role]:
return True
return False
def get_user_permissions(self, user_id: int) -> set:
"""获取用户所有权限"""
permissions = set()
if user_id in self.user_roles:
for role in self.user_roles[user_id]:
if role in self.role_permissions:
permissions.update(self.role_permissions[role])
return permissions
# 使用示例
rbac = RBACManager()
rbac.add_role('admin', ['create', 'read', 'update', 'delete'])
rbac.add_role('editor', ['create', 'read', 'update'])
rbac.add_role('viewer', ['read'])
rbac.assign_role(1, 'admin')
rbac.assign_role(2, 'editor')
多层权限校验封装
class PermissionChain:
"""权限链模式，实现多层权限校验"""
def __init__(self):
self.handlers = []
def add_handler(self, handler):
self.handlers.append(handler)
return self
def check(self, user, resource, action):
"""执行权限链校验"""
for handler in self.handlers:
result = handler.check(user, resource, action)
if not result:
return False
return True
class AuthenticationHandler:
"""认证处理器"""
def check(self, user, resource, action):
return user is not None and user.is_authenticated
class AuthorizationHandler:
"""授权处理器"""
def __init__(self, required_permission):
self.required_permission = required_permission
def check(self, user, resource, action):
return self.required_permission in user.permissions
class OwnershipHandler:
"""所有权处理器"""
def check(self, user, resource, action):
if action in ['read', 'update', 'delete']:
return resource.owner_id == user.id
return True
class RateLimitHandler:
"""频率限制处理器"""
def check(self, user, resource, action):
# 检查访问频率
pass
# 使用示例
permission_chain = PermissionChain()
permission_chain.add_handler(AuthenticationHandler())
permission_chain.add_handler(AuthorizationHandler('delete'))
permission_chain.add_handler(OwnershipHandler())
# 校验权限
user = get_current_user()
document = get_document(123)
if permission_chain.check(user, document, 'delete'):
# 执行删除操作
pass
实际应用示例
from flask import Flask, request, jsonify
from functools import wraps
import jwt
app = Flask(__name__)
# 权限中间件装饰器
def permission_required(*permissions):
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
# 获取token并解析用户信息
token = request.headers.get('Authorization')
if not token:
return jsonify({'error': '未认证'}), 401
try:
user_info = decode_jwt(token)
except:
return jsonify({'error': '无效的token'}), 401
# 创建权限管理器
permission_manager = PermissionManager(user_info)
# 校验权限
for permission in permissions:
if not permission_manager.has_permission(permission):
return jsonify({'error': '权限不足'}), 403
# 将用户信息注入到request中
request.user = user_info
return f(*args, **kwargs)
return decorated_function
return decorator
@app.route('/api/documents/<document_id>', methods=['DELETE'])
@permission_required('document:delete')
def delete_document(document_id):
"""删除文档接口"""
document = get_document(document_id)
# 二次校验：检查资源所有权
if document.owner_id != request.user.id:
# 检查是否为管理员
if 'admin' not in request.user.roles:
return jsonify({'error': '无权删除该文档'}), 403
# 执行删除
document.delete()
return jsonify({'message': '删除成功'})
# 用户上下文管理
class UserContext:
"""用户上下文管理器"""
_thread_local = threading.local()
@classmethod
def set_current_user(cls, user):
cls._thread_local.current_user = user
@classmethod
def get_current_user(cls):
return getattr(cls._thread_local, 'current_user', None)
@classmethod
def clear(cls):
if hasattr(cls._thread_local, 'current_user'):
del cls._thread_local.current_user
# 上下文管理器使用
@app.before_request
def before_request():
"""请求前的权限上下文初始化"""
token = request.headers.get('Authorization')
if token:
user = decode_jwt(token)
UserContext.set_current_user(user)
@app.after_request
def after_request(response):
"""请求后的清理"""
UserContext.clear()
return response
最佳实践建议
配置化管理
# permissions_config.py
PERMISSIONS_CONFIG = {
'document:create': ['admin', 'editor'],
'document:read': ['admin', 'editor', 'viewer'],
'document:update': ['admin', 'editor'],
'document:delete': ['admin'],
'user:manage': ['admin'],
}
# 从配置加载权限
def load_permissions_from_config():
return PERMISSIONS_CONFIG
日志记录
import logging
from functools import wraps
def audit_log(func):
@wraps(func)
def wrapper(*args, **kwargs):
user = get_current_user()
logging.info(f"用户{user.id}执行了{func.__name__}")
try:
result = func(*args, **kwargs)
logging.info(f"操作成功")
return result
except PermissionDeniedError as e:
logging.warning(f"权限校验失败: {e}")
raise
return wrapper
缓存优化
from functools import lru_cache
class CachedPermissionManager:
@lru_cache(maxsize=100)
def has_permission(self, user_id, permission_name):
# 实际权限检查逻辑
pass
这些封装方案可以根据实际需求组合使用,建议：
- 使用多层校验机制
- 实现细粒度的权限控制
- 做好审计日志
- 考虑性能优化（缓存）
- 统一错误处理
