Python防越权案例如何封装权限防越权

wen python案例 2026-06-28 261

本文目录导读：

装饰器方式封装
类方法权限封装
RBAC（基于角色的访问控制）封装
多层权限校验封装
实际应用示例
最佳实践建议

我来介绍几种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

这些封装方案可以根据实际需求组合使用,建议：