Django REST+React TypeScript认证遇401错误求助
我正在用Django REST Framework和React TypeScript实现完整的用户认证功能,但一直出现401 Unauthorized错误,折腾了三周还没解决,求帮忙排查问题。相关代码如下:
Django部分
models.py
from django.db import models from django.contrib.auth.models import AbstractUser # Create your models here. class Users(AbstractUser): first_name = models.CharField(max_length=256) last_name = models.CharField(max_length=256) email = models.CharField(max_length=256,unique=True) password = models.CharField(max_length=256) username = None USERNAME_FIELD = 'email' REQUIRED_FIELDS = [] class UserToken(models.Model): user_id = models.IntegerField() token = models.CharField(max_length=256) created_at = models.DateTimeField(auto_now_add=True) expired_at = models.DateTimeField(auto_now=True) class Reset(models.Model): email = models.CharField(max_length=255) token = models.CharField(max_length=256)
views.py
from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import exceptions from .Serializers import UserSerializers from .models import Users, UserToken, Reset from .authentication import create_access_token, create_refresh_token, JWTAuthentication, decode_refresh_token import datetime, random, string from django.core.mail import send_mail class RegisterAPIView(APIView): def post(self, request): data = request.data if data['password'] != data['password_confirm']: raise exceptions.APIException('password do not match!') serializer = UserSerializers(data=data) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data) class LoginAPIView(APIView): def post(self, request): email = request.data['email'] password = request.data['password'] user = Users.objects.filter(email=email).first() if user is None: raise exceptions.AuthenticationFailed('Invalid credentials ') if not user.check_password(password): raise exceptions.AuthenticationFailed('Invalid credentials') access_token = create_access_token(user.id) refresh_token = create_refresh_token(user.id) UserToken.objects.create( user_id = user.id, token = refresh_token, expired_at = datetime.datetime.utcnow() + datetime.timedelta(days=7) ) response = Response() response.set_cookie(key='refresh_token', value=refresh_token, httponly=True) response.data = { 'token': access_token } return response class UserAPIView(APIView): authentication_classes = [JWTAuthentication] def get(self, request): return Response(UserSerializers(request.Users).data) class RefreshAPIView(APIView): def post(self, request): refresh_token = request.COOKIES.get('refresh_token') id = decode_refresh_token(refresh_token) if not UserToken.objects.filter( user_id =id, token = refresh_token, expired_at__gt = datetime.datetime.now(tz=datetime.timezone.utc) ).exists(): raise exceptions.AuthenticationFailed('unathenticated') access_token = create_access_token(id) return Response({ 'token': access_token }) class LogoutAPIView(APIView): def post(self, request): refresh_token = request.COOKIES.get('refresh_token') UserToken.objects.filter(token= refresh_token).delete() response = Response() response.delete_cookie(key='refresh_token') response.data = { 'message': 'success' } return response class ForgetAPIView(APIView): def post(self, request): email = request.data['email'] token = ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(10)) Reset.objects.create( email = email, token = token ) url = 'http//localhost:3000/reset/' + token send_mail( subject = 'Reset Your Password ', message = 'Click <a href="%s">Here</a> to reset your password!' % url, from_email = 'from@example.com', recipient_list = [email] ) return Response({ 'message': 'success' }) class ResetAPIView(APIView): def post(self, request): data = request.data if data['password'] != data['password_confirm']: raise exceptions.APIException('password do not match!') reset_password = Reset.objects.filter(token=data['token']).first() if not reset_password: raise exceptions.APIException('Invalid Link!') user = Users.objects.filter(email=reset_password.email).first() if not user: raise exceptions.APIException('User not found') user.set_password(data['password']) user.save() return Response({ 'message': 'success' })
authentication.py
import jwt, datetime, pytz from django.utils.timezone import utc from .models import Users from rest_framework import exceptions from rest_framework.authentication import BaseAuthentication, get_authorization_header class JWTAuthentication(BaseAuthentication): def authenticate(self, request): auth = get_authorization_header(request).split() if auth and len(auth) == 2: token = auth[1].decode('utf-8') id = decode_access_token(token) user = Users.objects.get(pk=id) return (user, None) raise exceptions.AuthenticationFailed('unAuthenticated') def create_access_token(id): return jwt.encode({ 'user_id': id, 'exp': datetime.datetime.utcnow() + datetime.timedelta(seconds=30), 'iat': datetime.datetime.utcnow() }, 'access_secret', algorithm='HS256') def decode_access_token(token): try: payload = jwt.decode(token, 'access_secret', algorithms='HS256') return payload['user_id'] except: raise exceptions.AuthenticationFailed('unAuthenticated') def create_refresh_token(id): return jwt.encode({ 'user_id': id, 'exp': datetime.datetime.utcnow() + datetime.timedelta(days=7), 'iat': datetime.datetime.utcnow().replace(tzinfo=pytz.utc) }, 'refresh_secret', algorithm='HS256') def decode_refresh_token(token): try: payload = jwt.decode(token, 'refresh_secret', algorithms='HS256') return payload['user_id'] except: raise exceptions.AuthenticationFailed('unAuthenticated')
exception.py
from rest_framework.views import exception_handler def status_code_handler(exc, context): response = exception_handler(exc, context) if response is not None and response.status_code == 403: response.status_code = 401 return response
serializers.py
from .models import Users from rest_framework.serializers import ModelSerializer class UserSerializers(ModelSerializer): class Meta: model = Users fields = ['id', 'first_name', 'last_name', 'email', 'password'] extra_kwaegs = { 'password' : {'write_only': True} } def create(self, validated_data): password = validated_data.pop('password', None) instance = self.Meta.model(**validated_data) if password is not None: instance.set_password(password) instance.save() return instance
React TypeScript部分
Home.tsx
import axios from 'axios'; import React, { useEffect, useState } from 'react' function Home() { const [message, setMessage] = useState(''); useEffect(() => { (async () => { try{ const response = await axios.get('user'); const user = response.data; setMessage(`Hello ${user.first_name}`); }catch (e){ setMessage('Your Are Not Authantication'); } })(); }, []); return ( <div className='container mt-5 text-center text-white'> <h3>{message}</h3> </div> ) } export default Home
Axios.ts
import axios from "axios"; axios.defaults.baseURL = 'http://localhost:8000/'; axios.defaults.withCredentials = true; axios.interceptors.response.use(resp => resp, async error => { if(error.response.status === 401) { const response = await axios.post('refresh', {}); if(response.status === 200){ return axios(error.config) } } return error; }) ;
核心问题排查点
Django端问题
- UserAPIView属性错误:
request.Users应为request.user,DRF认证后用户对象存储在request.user属性中,大写的Users是模型类,会直接抛出属性不存在的错误,触发401。 - Serializer配置拼写错误:
extra_kwaegs应为extra_kwargs,该错误会导致密码的write_only配置失效,属于代码隐患。 - CORS配置缺失:跨域请求携带Cookie需要Django配置CORS支持,安装
django-cors-headers后在settings.py添加:INSTALLED_APPS = [ # ... 'corsheaders', ] MIDDLEWARE = [ 'corsheaders.middleware.CorsMiddleware', 'django.middleware.common.CommonMiddleware', # ... ] CORS_ALLOW_CREDENTIALS = True CORS_ALLOWED_ORIGINS = [ "http://localhost:3000", ] - RefreshToken过期时间被覆盖:
UserToken模型的expired_at字段使用auto_now=True,会在创建时自动覆盖为当前时间,导致刚生成的token直接过期,需改为expired_at = models.DateTimeField(),手动设置过期时间。
React端问题
- 请求未携带Authorization头:登录后获取的access token需存储到
localStorage,并通过请求拦截器添加到请求头:// 登录成功后保存token const loginRes = await axios.post('login', { email, password }); localStorage.setItem('access_token', loginRes.data.token); // 请求拦截器添加认证头 axios.interceptors.request.use(config => { const token = localStorage.getItem('access_token'); if (token) { config.headers.Authorization = `Bearer ${token}`; } return config; }); - RefreshToken更新后未同步本地存储:拦截器获取新access token后,需更新
localStorage中的值,并修改原请求的认证头,否则重试请求仍使用无效token:axios.interceptors.response.use(resp => resp, async error => { if(error.response.status === 401) { try { const refreshRes = await axios.post('refresh', {}); const newToken = refreshRes.data.token; localStorage.setItem('access_token', newToken); error.config.headers.Authorization = `Bearer ${newToken}`; return axios(error.config); } catch { window.location.href = '/login'; return Promise.reject(error); } } return Promise.reject(error); });
内容的提问来源于stack exchange,提问作者Faizan Qureshi
相关产品推荐
相关产品推荐

