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

🕗 发布于 2024-09-25 03:44 django 数据库

一、项目创建并实现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, include


urlpatterns = [
    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 JWTAuthentication



class 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 BlogView

urlpatterns = [
    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 ClientSession


class 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 BasePermission

class 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 functools

def 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_view

    return 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一分钟。

原文地址：https://blog.csdn.net/2201_75632987/article/details/142500448

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：9.24 C++ 常成员，运算符重载
下一篇：时间序列LSTM实现