返回首页

NextAuth + Django JWT:统一认证

本文描述了 NextAuth 与 Django JWT 的集成,实现统一认证而无需重复登录。它涵盖了自定义用户模型、credentials provider、OAuth 同步以及会话优化。提供完整的工作代码。

NextAuth 与 Django JWT,无令牌混乱
Advertisement 728x90

NextAuth + Django JWT:无缝单点登录,无重复认证

Next.js 中的 NextAuth 与 Django REST Framework 使用 JWT 时,认证流程常常冲突。用户在前台只需登录一次,即可从 Django 获取访问令牌和刷新令牌,所有 API 调用使用单一客户端,并自动刷新令牌。OAuth 提供商与后端同步——无需二次登录页面。

架构职责分离

NextAuth 负责 UI 会话、登录表单和 OAuth。Django 管理用户领域模型、JWT 颁发,以及通过 JWTAuthentication 保护 API。前端从不自行生成令牌——Django 是 API 访问的唯一来源。令牌存储在 NextAuth 会话中。

这种架构的关键优势:

Google AdInline article slot
  • 无需手动使用 localStorage 存储访问令牌
  • 自动刷新,无 401 错误
  • OAuth 同步,无重复登录
  • 单一 axios 客户端处理所有请求

Django:以邮箱为主的用户模型和自定义 JWT

用户模型将邮箱设为 USERNAME_FIELD。这简化了 OAuth 集成,因为用户名并非总是可用。

# auth_app/models.py
from django.contrib.auth.models import AbstractUser
from django.db import models
from django.contrib.auth.models import BaseUserManager

class CustomUserManager(BaseUserManager):
    def create_user(self, email, password=None, **extra_fields):
        if not email:
            raise ValueError("邮箱字段必须设置")
        email = self.normalize_email(email)
        user = self.model(email=email, **extra_fields)
        user.set_password(password)
        user.save(using=self._db)
        return user

    def create_superuser(self, email, password=None, **extra_fields):
        extra_fields.setdefault("is_staff", True)
        extra_fields.setdefault("is_superuser", True)
        return self.create_user(email, password, **extra_fields)


class User(AbstractUser):
    email = models.EmailField(unique=True)
    name = models.CharField(max_length=100, blank=True)
    provider = models.CharField(max_length=50, default="credentials", blank=True, null=True)

    USERNAME_FIELD = "email"
    REQUIRED_FIELDS = []

    objects = CustomUserManager()

settings.py 中:AUTH_USER_MODEL = "auth_app.User"。安装 rest_framework_simplejwtdj_rest_authallauth

自定义 TokenObtainPairSerializer,在令牌中添加 emailnameprovider

Google AdInline article slot
# auth_app/serializers.py
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer

class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):
    @classmethod
    def get_token(cls, user):
        token = super().get_token(user)
        token["email"] = user.email
        token["name"] = user.name
        token["provider"] = user.provider
        return token

    def validate(self, attrs):
        credentials = {
            "email": attrs.get("email"),
            "password": attrs.get("password"),
        }
        return super().validate(credentials)

登录视图检查邮箱验证,并返回令牌 + 用户数据:

# auth_app/views.py
class CustomTokenObtainPairView(TokenObtainPairView):
    serializer_class = CustomTokenObtainPairSerializer

    def post(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        user = serializer.user
        email_address = EmailAddress.objects.filter(user=user, email=user.email).first()
        if not email_address or not email_address.verified:
            raise AuthenticationFailed(detail="邮箱地址未验证。")

        return Response(
            {
                "access": serializer.validated_data["access"],
                "refresh": serializer.validated_data["refresh"],
                "user": {
                    "email": user.email,
                    "name": user.name,
                    "id": user.id,
                },
            },
            status=status.HTTP_200_OK,
        )

Next.js:凭证提供商作为 Django 代理

凭证提供商将邮箱/密码转发至 Django,并将收到的令牌存储在会话中:

// src/lib/auth/authOptions.ts
import { NextAuthOptions } from "next-auth";
import GoogleProvider from "next-auth/providers/google";
import CredentialsProvider from "next-auth/providers/credentials";
import apiClient from "@/services/authClientService";

const baseURL = process.env.NEXT_PUBLIC_API_BASE_URL;

export const authOptions: NextAuthOptions = {
  providers: [
    GoogleProvider({
      clientId: process.env.GOOGLE_CLIENT_ID as string,
      clientSecret: process.env.GOOGLE_CLIENT_SECRET as string,
    }),
    CredentialsProvider({
      name: "凭证登录",
      credentials: {
        email: { label: "邮箱", type: "text" },
        password: { label: "密码", type: "password" },
      },
      async authorize(credentials) {
        if (!credentials?.email || !credentials.password) {
          throw new Error("必须提供邮箱和密码");
        }

        const { data } = await apiClient.post(`${baseURL}/api/auth/custom/login/`, {
          email: credentials.email,
          password: credentials.password,
        });

        if (data?.user) {
          const { id, name, email } = data.user;
          return {
            id: id.toString(),
            name,
            email,
            accessToken: data.access || null,
            refreshToken: data.refresh || null,
          };
        }

        return null;
      },
    }),
  ],
};

OAuth 同步:注册或登录端点

Google OAuth 后,调用 register-or-login 端点。Django 创建或更新用户并颁发 JWT:

Google AdInline article slot
callbacks: {
  async signIn({ user, account }) {
    const allowedProviders = ["google", "apple"];

    if (account?.provider && allowedProviders.includes(account.provider)) {
      const res = await apiClient.post(
        `${baseURL}/api/auth/custom/oauth/register-or-login/`,
        {
          id: user.id,
          name: user.name,
          email: user.email,
          provider: account.provider,
        }
      );

      if (res.status === 200) {
        user.accessToken = res.data.access;
        user.refreshToken = res.data.refresh;
        return true;
      }

      return false;
    }

    return true;
  },
},

Django 端点:

class CustomOAuthRegisterOrLoginView(APIView):
    permission_classes = [AllowAny]

    def post(self, request, *args, **kwargs):
        serializer = OAuthUserSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        data = serializer.validated_data

        provider = data["provider"]
        email = data["email"]
        name = data["name"]
        user_id = data["id"]

        user, created = User.objects.get_or_create(
            email=email,
            defaults={
                "name": name,
                "provider": provider,
                "username": user_id,
            },
        )

        if not created:
            user.name = name
            user.provider = provider
            user.save()

        refresh = RefreshToken.for_user(user)
        access = str(refresh.access_token)

        return Response(
            {
                "message": "用户同步成功。",
                "user": {
                    "email": user.email,
                    "name": user.name,
                    "provider": user.provider,
                },
                "access": access,
                "refresh": str(refresh),
            },
            status=status.HTTP_200_OK,
        )

JWT 和会话回调优化

JWT 和会话仅存储 idaccessTokenrefreshToken。排除额外数据以最小化体积:

async jwt({ token, user }) {
  if (user) {
    token.id = user.id.toString();
    token.accessToken = user.accessToken || null;
    token.refreshToken = user.refreshToken || null;
  }
  return token;
},

async session({ session, token }) {
  session.user = {
    ...session.user,
    id: token.id as string,
  };
  session.accessToken = token.accessToken || null;
  session.refreshToken = token.refreshToken || null;
  return session;
},

核心要点

  • 单一令牌来源:Django JWT 是唯一 API 访问权威
  • 自动 OAuth 同步:注册或登录端点消除重复登录
  • 精简会话:仅 id + 令牌,无冗余
  • 邮箱验证:内置于登录流程,确保一致性
  • Axios 拦截器:自动刷新,401 时自动登出

— Editorial Team

Advertisement 728x90

继续阅读