当前位置：首页 > ds >正文

Django RBAC项目后端实战 - 03 DRF权限控制实现

ds 2025/7/18 23:20:19

项目背景

在上一篇文章中，我们完成了JWT认证系统的集成。本篇文章将实现基于Redis的RBAC权限控制系统，为系统提供细粒度的权限控制。

开发目标

实现基于Redis的权限缓存机制
开发DRF权限控制类
实现权限管理API
配置权限白名单

前置配置

在开始开发权限校验相关代码之前，需要先在 settings.py 中完成以下配置：

1. Redis配置

CACHES = {"default": {"BACKEND": "django_redis.cache.RedisCache","LOCATION": "redis://127.0.0.1:6379/1",  # 需要根据实际情况修改ip和端口"OPTIONS": {"CLIENT_CLASS": "django_redis.client.DefaultClient",}}
}# 权限缓存配置
PERMISSION_CACHE_TIMEOUT = 3600  # 权限缓存过期时间（单位：秒），默认1小时

2. 权限白名单配置

# 权限白名单配置
PERMISSION_WHITELIST = ['/api/auth/login/','/api/auth/register/','/api/auth/refresh/',
]

权限缓存工具类

在rbac/utils.py中创建权限缓存工具类：

from django.core.cache import cache
from django.conf import settings
from .models import Permission
from django.contrib.auth import get_user_modelUser = get_user_model()class PermissionCache:"""权限缓存工具类用于管理用户权限的缓存操作，包括获取和清除缓存"""@staticmethoddef get_user_permissions(user_id):"""获取用户的权限列表优先从缓存中获取，如果缓存不存在则从数据库查询并缓存Args:user_id: 用户IDReturns:list: 用户权限列表，包含权限的codename"""# 构建缓存键cache_key = f"user_permissions_{user_id}"try:# 尝试从缓存获取权限permissions = cache.get(cache_key)# 如果缓存中没有，从数据库查询if permissions is None:permissions = PermissionCache._get_permissions_from_db(user_id)# 将查询结果存入缓存try:cache.set(cache_key, list(permissions), settings.PERMISSION_CACHE_TIMEOUT)except Exception:# 如果缓存操作失败，忽略错误继续执行passreturn permissionsexcept Exception:# 发生异常时从数据库查询return PermissionCache._get_permissions_from_db(user_id)@staticmethoddef _get_permissions_from_db(user_id):"""从数据库直接查询用户权限"""try:# 方式一：使用查询集user = User.objects.get(id=user_id)role_ids = user.roles.values_list('id', flat=True)permissions = Permission.objects.filter(role__in=role_ids).distinct().values_list('codename', flat=True)return list(permissions)except Exception:return []@staticmethoddef clear_user_permissions(user_id=None):"""清除用户权限缓存可以清除单个用户的缓存，也可以清除所有用户的缓存Args:user_id: 可选参数，指定要清除缓存的用户ID如果不传，则清除所有用户的缓存"""try:if user_id:# 清除指定用户的缓存cache.delete(f"user_permissions_{user_id}")else:# 清除所有用户的缓存（使用通配符）cache.delete_pattern("user_permissions_*")except Exception:# 如果清除缓存失败，忽略错误pass

权限校验机制实现

1. 权限校验流程

2. 权限缓存设计

为了提高权限校验的性能，我们使用Redis缓存用户权限。缓存设计如下：

缓存键格式：user_permissions_{user_id}
缓存时间：1小时

3. 核心代码实现

3.1 权限类完整实现

在rbac/permissions.py中实现权限校验类：

from rest_framework.permissions import BasePermission
from django.conf import settings
from .utils import PermissionCacheclass RBACPermission(BasePermission):def has_permission(self, request, view):# 1. 检查白名单if self._is_whitelist_path(request.path_info):return True# 2. 检查超级管理员if request.user.is_superuser:return True# 3. 获取权限标识permission_codename = self._get_permission_codename(request)# 4. 获取用户权限user_permissions = PermissionCache.get_user_permissions(request.user.id)# 5. 检查权限return permission_codename in user_permissionsdef _is_whitelist_path(self, path):return any(path.startswith(p) for p in settings.PERMISSION_WHITELIST)def _get_permission_codename(self, request):method = request.method.lower()path = request.path_inforeturn f"{method}:{path}"

注意：RBACPermission是一个标准的DRF权限类(Permission Class)，继承自BasePermission。我们将其放在专门的permissions.py文件中，符合DRF的最佳实践。

3.1.1 权限类的配置和使用

有两种方式可以使用RBACPermission权限类：

全局配置：在settings.py中为所有DRF视图设置默认权限类

# settings.py
REST_FRAMEWORK = {'DEFAULT_PERMISSION_CLASSES': ['rbac.permissions.RBACPermission','rest_framework.permissions.IsAuthenticated',],# 其他DRF配置...
}

注意：如果在全局配置了权限类，视图中就不需要重复设置相同的权限类了。只有在需要覆盖全局配置时，才需要在视图中设置permission_classes。

视图级别配置：在需要特殊权限控制的视图或视图集中设置权限类

# 在视图集中使用 (仅在需要覆盖全局配置时)
from rbac.permissions import RBACPermissionclass SpecialViewSet(viewsets.ModelViewSet):permission_classes = [IsAdminUser]  # 覆盖全局权限类配置queryset = SpecialModel.objects.all()serializer_class = SpecialSerializer# 视图集方法...

或者：

# 在功能视图中使用 (仅在需要覆盖全局配置时)
from rbac.permissions import RBACPermission
from rest_framework.decorators import api_view, permission_classes@api_view(['GET'])
@permission_classes([IsAdminUser])  # 覆盖全局权限类配置
def special_view(request):# 视图代码...return Response(...)

3.2 视图集实现

3.2.1 权限管理视图集

在rbac/views.py中实现权限管理视图集：

from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from django.core.cache import cache
from .models import Permission, Role
from .serializers import PermissionSerializer
from .utils import PermissionCache
from .permissions import RBACPermission  class PermissionViewSet(viewsets.ModelViewSet):"""权限管理视图集提供权限的增删改查，并在权限变更时自动清除缓存"""queryset = Permission.objects.all()serializer_class = PermissionSerializerdef perform_create(self, serializer):"""创建权限后清除所有用户的权限缓存"""serializer.save()PermissionCache.clear_user_permissions()def perform_update(self, serializer):"""更新权限后清除所有用户的权限缓存"""serializer.save()PermissionCache.clear_user_permissions()def perform_destroy(self, instance):"""删除权限后清除所有用户的权限缓存"""instance.delete()PermissionCache.clear_user_permissions()

3.2.2 角色管理视图集

在rbac/views.py中实现角色管理视图集：

from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
from django.core.cache import cache
from .models import Role, Permission
from .serializers import RoleSerializer
from .permissions import RBACPermission  class RoleViewSet(viewsets.ModelViewSet):"""角色管理视图集提供角色的增删改查，并在角色变更时自动清除缓存"""queryset = Role.objects.all()serializer_class = RoleSerializerdef perform_create(self, serializer):"""创建角色后清除所有用户的权限缓存"""serializer.save()PermissionCache.clear_user_permissions()def perform_update(self, serializer):"""更新角色后清除所有用户的权限缓存"""serializer.save()PermissionCache.clear_user_permissions()def perform_destroy(self, instance):"""删除角色后清除所有用户的权限缓存"""instance.delete()PermissionCache.clear_user_permissions()@action(detail=True, methods=['post'])def assign_permissions(self, request, pk=None):"""为角色分配权限"""role = self.get_object()permission_ids = request.data.get('permission_ids', [])# 获取权限对象permissions = Permission.objects.filter(id__in=permission_ids)# 更新角色的权限role.permissions.set(permissions)PermissionCache.clear_user_permissions()return Response({"message": "权限分配成功","role": RoleSerializer(role).data})

3.3 路由配置

完成视图集的实现后，还需要配置URL路由才能通过API访问这些视图集。在rbac/urls.py中添加以下配置：

from django.urls import path, include
from rest_framework.routers import DefaultRouter
from . import views# 创建路由器
router = DefaultRouter()# 注册视图集，指定basename参数确保URL名称唯一
router.register(r'permissions', views.PermissionViewSet, basename='permission')
router.register(r'roles', views.RoleViewSet, basename='role')# 设置应用命名空间
app_name = 'rbac'# URL配置
urlpatterns = [path('rbac/', include(router.urls)),
]

然后，在项目的主urls.py文件中包含这些路由：

from django.urls import path, includeurlpatterns = [# 其他URL配置...path('api/rbac/', include('rbac.urls')),  # 将所有RBAC相关路由放在/api/rbac/前缀下
]

这样配置后，系统会自动生成以下API路由：

权限管理:

GET /api/rbac/permissions/ - 列出所有权限 (URL名称: rbac:permission-list)
POST /api/rbac/permissions/ - 创建新权限 (URL名称: rbac:permission-create)
GET /api/rbac/permissions/{id}/ - 获取指定权限 (URL名称: rbac:permission-detail)
PUT /api/rbac/permissions/{id}/ - 更新指定权限 (URL名称: rbac:permission-update)
DELETE /api/rbac/permissions/{id}/ - 删除指定权限 (URL名称: rbac:permission-destroy)

角色管理:

GET /api/rbac/roles/ - 列出所有角色 (URL名称: rbac:role-list)
POST /api/rbac/roles/ - 创建新角色 (URL名称: rbac:role-create)
GET /api/rbac/roles/{id}/ - 获取指定角色 (URL名称: rbac:role-detail)
PUT /api/rbac/roles/{id}/ - 更新指定角色 (URL名称: rbac:role-update)
DELETE /api/rbac/roles/{id}/ - 删除指定角色 (URL名称: rbac:role-destroy)
POST /api/rbac/roles/{id}/assign_permissions/ - 为角色分配权限 (URL名称: rbac:role-assign-permissions)

使用命名空间和basename参数的主要优势是：

避免不同应用中的URL名称冲突
在模板中可以使用 {% url 'rbac:permission-list' %}引用URL
在代码中可以使用reverse('rbac:role-detail', kwargs={'pk': role_id})生成URL

配置完路由后，可以在浏览器中访问这些API端点进行测试。如果您使用DefaultRouter，它还会自动生成可浏览的API文档页面，方便开发调试。

权限系统测试

在实现完DRF权限控制和权限管理API后，需要进行全面的测试以确保系统稳定可靠。我们采用分层测试策略，包括单元测试和接口测试两个层面。

测试策略概述

权限系统是整个应用的核心安全机制，需要特别全面的测试覆盖。我们的测试策略如下：

单元测试：测试代码内部逻辑和数据库操作
- 权限判断逻辑测试
- 数据库操作测试
- 缓存操作测试
- 权限类功能测试
接口测试：测试接口的可用性和权限控制
- 权限校验流程测试
- 权限管理流程测试

这种分层测试策略能够确保我们从代码内部逻辑到实际接口行为都进行了全面覆盖，提高系统的可靠性和安全性。

1. 单元测试

单元测试主要针对权限系统的内部逻辑和数据库操作进行测试，确保每个功能单元正常工作。我们使用Django的测试框架和APIClient进行单元测试。

1.1 权限判断逻辑测试

这部分测试主要验证权限校验中间件的核心逻辑是否正确。

def test_user_permissions(self):"""测试权限判断逻辑：1. 未授权用户应该无法访问受保护资源2. 授权用户应该能够访问已获权限资源3. 授权用户不能访问未获权限资源"""# 登录获取tokenresponse = self.client.post('/api/auth/login/', {'username': 'test_user','password': 'test123456'})token = response.data['data']['access']self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}')# 测试未分配权限时访问（应该失败）response = self.client.get('/api/rbac/permissions/')self.assertEqual(response.status_code, 403)# 分配权限self.user_role.permissions.add(self.view_permission)# 清除缓存以确保权限更新生效PermissionCache.clear_user_permissions(self.user.id)# 测试分配权限后访问（应该成功）response = self.client.get('/api/rbac/permissions/')self.assertEqual(response.status_code, 200)# 测试访问未授权资源（应该失败）response = self.client.post('/api/rbac/permissions/', {'codename': 'test:permission','desc': '测试权限'})self.assertEqual(response.status_code, 403)

1.2 管理员权限测试

测试管理员权限：

超级管理员应该能够访问所有资源，无需额外授权

def test_admin_permissions(self):"""测试管理员权限：1. 超级管理员应该能够访问所有资源，无需额外授权"""# 登录获取tokenresponse = self.client.post('/api/auth/login/', {'username': 'admin_user','password': 'admin123456'})token = response.data['data']['access']self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}')# 测试查看权限列表（应该成功）response = self.client.get('/api/rbac/permissions/')self.assertEqual(response.status_code, 200)# 测试创建权限（应该成功）response = self.client.post('/api/rbac/permissions/', {'codename': 'test:permission','desc': '测试权限'})self.assertEqual(response.status_code, 201)# 验证权限确实被创建self.assertTrue(Permission.objects.filter(codename='test:permission').exists())

1.3 权限分配测试

测试角色权限分配：

验证权限分配操作是否正确反映在数据库中
测试异常情况下的权限分配（不存在的权限ID等）
测试特殊场景（空权限列表、重复权限）

def test_assign_permissions(self):"""测试角色权限分配：1. 验证权限分配操作是否正确反映在数据库中2. 测试异常情况下的权限分配（不存在的权限ID等）3. 测试特殊场景（空权限列表、重复权限）"""# 登录获取tokenresponse = self.client.post('/api/auth/login/',{'username':'admin_user','password':'admin123456'})token = response.data['data']['access']self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}')# 测试新权限response = self.client.post('/api/rbac/permissions/',{'codename':'test:permission2',  # 使用不同的codename避免冲突'desc':'测试权限2'})new_permission_id  = response.data['id']# 测试分配权限response = self.client.post(f'/api/rbac/roles/{self.user_role.id}/assign_permissions/',{'permission_ids':[self.view_permission.id, new_permission_id]},format='json'  # 添加这个参数确保正确序列化)self.assertEqual(response.status_code, 200)# 验证权限分配结果已写入数据库self.assertEqual(set(self.user_role.permissions.values_list('id', flat=True)),{self.view_permission.id, new_permission_id})# 测试普通用户分配权限（应该失败，验证权限控制）response = self.client.post('/api/auth/login/', {'username': 'test_user','password': 'test123456'})token = response.data['data']['access']self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}')       response = self.client.post(f'/api/rbac/roles/{self.user_role.id}/assign_permissions/', {'permission_ids': [self.view_permission.id]})self.assertEqual(response.status_code, 403)# 测试边界情况，分配不存在的权限IDself.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}')response = self.client.post(f'/api/rbac/roles/{self.user_role.id}/assign_permissions/', {'permission_ids': [99999]  # 不存在的权限ID})     self.assertEqual(response.status_code, 403)# 测试边界情况，分配空权限列表self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}')response = self.client.post(f'/api/rbac/roles/{self.user_role.id}/assign_permissions/', {'permission_ids': []})self.assertEqual(response.status_code, 403)# 测试边界情况，分配重复权限self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}')response = self.client.post(f'/api/rbac/roles/{self.user_role.id}/assign_permissions/', {'permission_ids': [self.view_permission.id, self.view_permission.id]})self.assertEqual(response.status_code, 403)

1.4 缓存操作测试

测试权限缓存机制：

缓存创建：首次获取权限时应创建缓存
缓存读取：再次获取权限时应从缓存读取
缓存更新：权限变更后清除缓存，再次获取应包含新权限
缓存删除：清除缓存后应从数据库重新加载
批量缓存清除：应能清除多个用户的缓存
缓存过期：设置短暂过期时间后，缓存应自动失效

def test_permission_cache(self):"""测试权限缓存机制：1. 缓存创建：首次获取权限时应创建缓存2. 缓存读取：再次获取权限时应从缓存读取3. 缓存更新：权限变更后清除缓存，再次获取应包含新权限4. 缓存删除：清除缓存后应从数据库重新加载5. 批量缓存清除：应能清除多个用户的缓存6. 缓存过期：设置短暂过期时间后，缓存应自动失效"""# 登录获取tokenresponse = self.client.post('/api/auth/login/',{'username':'admin_user','password':'admin123456'})token = response.data['data']['access']self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}')# 1. 测试缓存创建# 分配权限给角色self.admin_role.permissions.add(self.view_permission)# 通过API请求触发缓存创建response = self.client.get('/api/rbac/permissions/')self.assertEqual(response.status_code, 200)# 验证缓存已创建并可以正确读取Permissions = PermissionCache.get_user_permissions(self.admin.id)self.assertIsNotNone(Permissions)# 2. 测试缓存更新# 创建新权限new_permission = Permission.objects.create(codename='test:permission3',desc='测试权限3')# 分配新权限给角色self.admin_role.permissions.add(new_permission)# 确保view_permission也被分配self.admin_role.permissions.add(self.view_permission)# 清除缓存PermissionCache.clear_user_permissions(self.admin.id)# 获取更新后的权限（应包含新权限）updated_permissions = PermissionCache.get_user_permissions(self.admin.id)self.assertEqual(updated_permissions, ['get:/api/rbac/permissions/', 'test:permission3'])# 3. 测试权限删除# 先从数据库中移除权限self.admin_role.permissions.remove(self.view_permission)# 清除缓存PermissionCache.clear_user_permissions(self.admin.id)# 验证权限已被移除permissions = PermissionCache.get_user_permissions(self.admin.id)self.assertEqual(permissions, ['test:permission3'])  # 现在应该只有test:permission3# 4.测试批量缓存清除# 为多个用户分配权限并创建缓存self.user_role.permissions.add(self.view_permission)PermissionCache.get_user_permissions(self.user.id)PermissionCache.get_user_permissions(self.admin.id)# 清除所有用户缓存PermissionCache.clear_user_permissions()# 验证所有用户缓存已被清除并重新加载user_permissions = PermissionCache.get_user_permissions(self.user.id)admin_permissions = PermissionCache.get_user_permissions(self.admin.id)self.assertEqual(user_permissions, ['get:/api/rbac/permissions/'])self.assertEqual(admin_permissions, ['test:permission3'])# 5.测试缓存过期# 设置缓存过期时间为1秒original_permissions = settings.PERMISSION_CACHE_TIMEOUTsettings.PERMISSION_CACHE_TIMEOUT = 1# 获取权限（创建缓存）PermissionCache.get_user_permissions(self.admin.id)# 验证缓存创建self.assertIsNotNone(PermissionCache.get_user_permissions(self.admin.id))# 等待缓存过期import timetime.sleep(2)# 验证缓存已过期并重新加载permissions = PermissionCache.get_user_permissions(self.admin.id)self.assertEqual(permissions, ['test:permission3'])# 恢复原始缓存时间settings.PERMISSION_CACHE_TIMEOUT = original_permissions

1.5 权限类功能测试

测试权限类特定功能：

白名单功能：白名单内的路径应无需权限即可访问
权限标识符生成：验证权限标识符生成逻辑正确性

def test_permission_class_functions(self):"""测试权限类特定功能：1. 白名单功能：白名单内的路径应无需权限即可访问2. 权限标识符生成：验证权限标识符生成逻辑正确性"""# 测试白名单功能# 白名单路径应直接通过权限检查response = self.client.get('/api/auth/login/')self.assertNotEqual(response.status_code, 403)# 测试权限标识符生成from rbac.permissions import RBACPermissionpermission_check = RBACPermission()# 创建模拟请求from django.http import HttpRequestrequest = HttpRequest()request.method = 'GET'request.path_info = '/api/rbac/permissions/'# 生成权限标识符permission_codename = permission_check._get_permission_codename(request)self.assertEqual(permission_codename, 'get:/api/rbac/permissions/')

2. 接口测试

接口测试主要验证权限系统在实际请求场景下的行为，确保接口能够正确响应不同权限的用户请求。我们使用ApiPost工具进行接口测试。

2.1 权限校验流程测试

接口测试的一个重要部分是验证权限校验流程，这包括以下场景：

未登录访问受保护接口
- 预期结果：返回 401 Unauthorized（未认证）
- 测试方法：不携带Token直接访问受保护接口
无权限访问受保护接口
- 预期结果：返回 403 Forbidden（禁止访问）
- 测试方法：使用普通用户Token访问未授权接口
有权限访问受保护接口
- 预期结果：返回 200 OK（请求成功）
- 测试方法：使用授权用户Token访问已授权接口

在ApiPost中设置以下测试场景：

# 场景1：未登录访问权限列表
GET /api/rbac/permissions/
Headers: {}  # 不携带认证Token# 场景2：无权限用户访问权限列表
GET /api/rbac/permissions/
Headers: {"Authorization": "Bearer <普通用户Token>"
}# 场景3：有权限用户访问权限列表
GET /api/rbac/permissions/
Headers: {"Authorization": "Bearer <已授权用户Token>"
}# 场景4：管理员访问权限列表
GET /api/rbac/permissions/
Headers: {"Authorization": "Bearer <管理员Token>"
}

2.2 权限管理流程测试

接口测试的另一个重要部分是验证权限管理流程，这包括以下场景：

创建权限
- 测试方法：使用管理员账户创建新权限
- 验证点：权限成功创建并可被查询
分配权限给角色
- 测试方法：将创建的权限分配给指定角色
- 验证点：角色成功获得权限
验证权限生效
- 测试方法：使用角色用户访问相应接口
- 验证点：用户能够成功访问已授权接口

在ApiPost中设置以下测试场景：

# 场景1：创建新权限
POST /api/rbac/permissions/
Headers: {"Authorization": "Bearer <管理员Token>"
}
Body: {"codename": "get:/api/users/","desc": "查看用户列表"
}# 场景2：分配权限给角色
POST /api/rbac/roles/{role_id}/assign_permissions/
Headers: {"Authorization": "Bearer <管理员Token>"
}
Body: {"permission_ids": [1, 2, 3]  # 权限ID列表
}# 场景3：验证权限生效
GET /api/users/  # 尝试访问刚授权的接口
Headers: {"Authorization": "Bearer <角色用户Token>"
}

3. 测试执行指南

3.1 运行单元测试

使用以下命令运行所有单元测试：

# 运行所有测试
python manage.py test rbac# 运行特定测试类
python manage.py test rbac.tests.RBACPermissionTest# 运行特定测试方法
python manage.py test rbac.tests.RBACPermissionTest.test_permission_cache