Django一分钟：DRF快速实现JWT认证与RBAC权限校验

一、项目创建并实现JWT认证

1. 下载依赖

下载django、djangorestframework、djangorestframework_simplejwt

pip install django djangorestframework djangorestframework_simplejwt

2. 创建项目

启动Django项目

django-admin startproject <myproject>
cd myproject

用你实际的项目名称替换<myproject>

创建app

python manage.py startapp <myapp>

用你实际的app名称替换<myapp>

myproject/
├── myproject/
|   ├── __init__.py
|   ├── settings.py
|   ├── urls.py
|   ├── wsgi.py
|   └── asgi.py
├── myapp/
│   ├── __init__.py
│   ├── admin.py
│   ├── apps.py
│   ├── migrations/
│   │   └── __init__.py
│   ├── models.py
│   ├── tests.py
│   ├── urls.py
│   └── views.py
└── manage.py

在配置文件中设置app

在settings.py文件中配置好需要的app

INSTALLED_APPS = [...'rest_framework','rest_framework_simplejwt','myapp',
]

3. 配置JWT

# settings.py
from datetime import timedelta # 添加在INSTALLED_APPS下
# 该配置用于指定默认使用的权限类和授权类
REST_FRAMEWORK = {'DEFAULT_PERMISSION_CLASSES': ('rest_framework.permissions.IsAuthenticated',),'DEFAULT_AUTHENTICATION_CLASSES': ('rest_framework_simplejwt.authentication.JWTAuthentication',),
}# 用于配置令牌过期时间等参数
SIMPLE_JWT = {'ACCESS_TOKEN_LIFETIME': timedelta(minutes=60),'SLIDING_TOKEN_REFRESH_LIFETIME': timedelta(days=1),'SLIDING_TOKEN_LIFETIME': timedelta(days=30),'SLIDING_TOKEN_REFRESH_LIFETIME_LATE_USER': timedelta(days=1),'SLIDING_TOKEN_LIFETIME_LATE_USER': timedelta(days=30),
}

4. 配置路由

from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView
from django.contrib import admin
from django.urls import path, includeurlpatterns = [path('admin/', admin.site.urls),path('api/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),    path('', include('myapp.urls')), 
]

5. 创建视图类并配置授权

创建视图类，并为视图类添加权限要求，这里我们先添加基本的授权要求，即要求用户必须在请求头中携带我们的JWT token才能访问相应的路径。

from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework.permissions import IsAuthenticated
from rest_framework_simplejwt.authentication import JWTAuthenticationclass BlogView(APIView):permission_classes = (IsAuthenticated,)authentication_classes = (JWTAuthentication,)def get(self, request, *args, **kwargs):return Response({'msg': 'success', 'detail': 'myblog'})

未新创建的视图创建urls.py文件

touch myapp/urls.py

创建完成后添加配置

from django.urls import path
from myapp.views import BlogViewurlpatterns = [path('blog/', BlogView.as_view(), name='blog'),
]

6. 启动项目

数据库迁移

python manage.py migrate

启动项目

python manage.py createsuperuser
python manage.py runserver

二、测试JWT

在上一节的最后我们创建了一个管理员账户，假如为：

username: admin
password: admin123

请求授权接口的方法：

使用djangorestframework_simplejwt创建的token接口，请求token时要求我们使用POST方法，并在请求体中携带用户名和密码：{"username": "admin", "password": "admin123"}。
如果请求成功，该接口会返回两个token，一个是access_token，另一个是refresh_token。当请求需要授权权限的接口时，需要在请求头中携带access_token。
access_token的存活时间较短，refresh token的存活时间长，access_token过期需要获取新令牌，获取新令牌需要携带在请求头中携带refresh_token向../refresh/token端口进行请求。
所谓携带token指的是在请求头中添加Authorization字段，具体的格式时Authorization: Bearer <token>，在代码中体现为{"Authorization": f"Bearer {token}"}。

现在我们需要一个客户端来测试我们创建的后端服务，你可以通过postman创建测试请求，也可以通过http标准库、requests或aiohttp创建客户端进行测试，下面以aiohttp为例：

import asyncio
from aiohttp import ClientSessionclass Client:"""测试客户端"""def __init__(self):self.url = "http://localhost:8000/"self.user = {"username": "admin", "password": "admin123"}self.session = ClientSession()self.access_token = ""self.refresh_token = ""async def close(self):await self.session.close()async def get_token(self):"""获取token"""url = self.url + "auth/token/"async with self.session.post(url, json=self.user) as response:if response.status == 200:data = await response.json()if "access" in data and "refresh" in data:self.access_token = data["access"]self.refresh_token = data["refresh"]print(f"access_token: {self.access_token}")print(f"refresh_token: {self.refresh_token}")else:data.update({"error": "fail to get token"})print(data)else:print(f"Error status code: {response.status}")async def refresh_token(self):"""刷新token"""url = self.url + "auth/token/refresh/"headers = {"Authorization": f"Bearer {self.refresh_token}"}async with self.session.post(url, headers=headers) as response:if response.status == 200:data = await response.json()if "access" in data:self.access_token = data["access"]print(f"access_token: {self.access_token}")else:data.update({"error": "fail to refresh token"})print(data)else:print(f"Error status code: {response.status}")async def get_blog(self):"""获取博客"""url = self.url + "blog/"headers = {"Authorization": f"Bearer {self.access_token}"}async with self.session.get(url, headers=headers) as response:if response.status == 200:print(await response.json())else:print(f"Error status code: {response.status}")async def main():client = Client()await client.get_token()await client.get_blog()await client.close()if __name__ == "__main__":asyncio.run(main())

在前端项目中对接该接口，需要使用axios或fetch发起请求，可以把获取到的token存贮在localStorage中，每次请求时携带授权请求头。

三、权限分配与验证

1. Django Auth基础知识

在本文中我们将使用Django自带的auth系统来实现RBAC权限校验，在此之前需要了解一些关于Djangoauth系统的基础知识。

注册Django的auth应用，在初次进行migrate数据库迁移的时候，Django会自动在数据库中创建5张表：用户、权限、组以及三者两两之间的关系表。这在RBAC权限管理系统的数据库表设计中非常常见。

user
group
permission
user_group
group_permission
user_permission

在使用Django的认证系统我们需要知道以下几件事：

我们可以自己在permission表中创建一些权限，但通常来说不需要，Django在执行数据库迁移时，会自动为已注册app的模型创建增、删、改、查四个权限。
我们可以为用户分配权限，本质上就是在user_permission关系表中创建一条数据。我们也可以创建一个组，你可以将组命名为“采购部门”，为组分配权限，被分配到这个组中的用户将自动获取这个组的权限。
通过createsuperuser 创建的超级用户会拥有所有的权限(准确来说是自动通过权限认证)，普通用户的权限需要自己分配。

2，为用户分配权限

打开Django的shell控制台：

python manage.py shell

创建测试用户：

from django.contrib.auth.models import User
User.objects.create_user(username="test", email="test@qq.com",password="test123")

为新创建的用户分配权限view_blog：

from django.contrib.auth.models import Permission
permission = Permission.objects.get(codename="view_blog")
user = User.objects.get(username="test")
user.user_permissions.add(permission)
user.save()

此时你可以通过一些数据库工具查询到，你的sqlite数据库中的user_permission关系表中新增了一条数据，这就表示我们为test用户分配了view_blog权限。

3. 为路视图类添加权限

我们可以通过drf自定义权限类的方式为APIView整体添加权限限制：

from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework.permissions import IsAuthenticated
from rest_framework_simplejwt.authentication import JWTAuthentication
from apps.authorization.decorators import class_permission_required
from rest_framework.permissions import BasePermissionclass ViewBlogP(BasePermission):def has_permission(self, request, view):return request.user.has_perm('auth.view_user')class BlogView(APIView):permission_classes = (IsAuthenticated, ViewBlogP)authentication_classes = (JWTAuthentication,)def get(self, request, *args, **kwargs):return Response({'msg': 'success', 'detail': 'myblog'})

4.为视图类方法添加权限

django的permission_required装饰器可以为视图方法创建权限要求，不过permission_required不能直接在视图类的方法上直接使用，我们需要创建一个适配装饰器如下，你可以放置在utils.py文件中：

# utils.py
from django.contrib.auth.decorators import permission_required
import functoolsdef class_permission_required(perm, login_url=None, raise_exception=False):"""适配装饰器使得permission_required装饰器在视图类的成员方法上也能使用"""original_decorator = permission_required(perm, login_url, raise_exception)def adapter(view_method):@functools.wraps(view_method)def wrapped_view(self, request, *args, **kwargs):def new_func(request, *args, **kwargs):return view_method(self, request, *args, **kwargs)decorated_func = original_decorator(new_func)return decorated_func(request, *args, **kwargs)return wrapped_viewreturn adapter

使用方法：

class BlogView(APIView):permission_classes = (IsAuthenticated,)authentication_classes = (JWTAuthentication,)@class_permission_required('auth.view_user', raise_exception=True)def get(self, request, *args, **kwargs):return Response({'msg': 'success'})

关于此方法的更多细节请参考我的另一篇文章，欢迎订阅我的免费专栏Django一分钟。