Django初窥门径-oauth登录认证

引言

在现代Web应用程序中，用户身份验证和授权是至关重要的组成部分。Django，一个流行的Python Web框架，为用户身份验证提供了内置支持。本文将探讨如何创建和注册Django应用，自定义身份验证服务，配置AUTHENTICATION_BACKENDS，创建视图以及如何使用API视图获取当前登录用户的信息。

Django框架和OAuth身份验证

Django是一个高度可定制的Web框架，它提供了强大的工具和库，用于快速构建Web应用程序。OAuth是一种开放标准，用于安全地授权访问资源，通常用于身份验证。结合Django和OAuth，您可以构建安全和可扩展的用户身份验证和授权系统。

创建应用和注册应用

要创建一个Django应用，您可以使用以下命令：

python manage.py startapp oauth

然后，在项目的settings.py文件中，将应用注册到INSTALLED_APPS列表中，如下所示：

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',
    'drf_yasg2',
    'django_filters',
    'account.apps.AccountConfig',
    'oauth'
]

身份验证服务

创建自定义身份验证服务

#!/usr/bin/python
# -*- coding: utf-8 -*-
from django.contrib.auth import get_user_model
from django.contrib.auth.backends import ModelBackend
from django.core.cache import cache
from django.db.models import Q
from django.http import Http404
from django.shortcuts import get_object_or_404
from django.utils import timezone
from rest_framework import exceptions
from rest_framework.authentication import TokenAuthentication as BaseTokenAuthentication

User = get_user_model()


def get_ip(request):
    """获取当前请求ip
    :param request: 当前请求上下文
    :return: ip地址
    """
    x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
    if x_forwarded_for:
        ip = x_forwarded_for.split(',')[0]
    else:
        ip = request.META.get('REMOTE_ADDR')
    return ip


class EmailOrUsernameModelBackend(ModelBackend):
    """自定义用户验证
    允许用户使用用户名或者邮箱登录
    允许使用密码或者邮箱验证码校验
    """

    def authenticate(self, request, username=None, password=None, login_type='account', **kwargs):
        try:
            user = User.objects.get(Q(username=username) | Q(email=username), is_active=True)
            if login_type == 'account' and user.check_password(password):
                user.last_login = timezone.now()
                user.last_login_ip = get_ip(request)
                user.save()
                return user
            else:
                # 邮箱验证码校验
                code = cache.get('account:email:{0}:login:code'.format(user.email))
                if password == code:
                    return user
                return None

        except User.DoesNotExist:
            return None


class TokenAuthentication(BaseTokenAuthentication):
    """token认证"""
    keyword = 'bearer'

    def authenticate_credentials(self, key):
        token_user_cache_key = f'oauth:token:{key}:user:id'
        if cache.ttl(token_user_cache_key) == 0:
            raise exceptions.AuthenticationFailed('token无效')
        try:
            user_id = cache.get(token_user_cache_key)
            user_token_cache_key = f'oauth:user:id:{user_id}:token'
            if cache.ttl(user_token_cache_key) != 0 and cache.get(user_token_cache_key) == key:
                user = get_object_or_404(User, id=user_id, is_active=True)
                return user, key
            raise exceptions.AuthenticationFailed('token错误')
        except Http404:
            raise exceptions.AuthenticationFailed('token无效')

在自定义身份验证服务中，我们创建了EmailOrUsernameModelBackend和TokenAuthentication类，以改进用户身份验证和安全性。EmailOrUsernameModelBackend允许用户使用用户名或邮箱登录，还支持邮箱验证码校验。TokenAuthentication用于验证用户的令牌。这些自定义服务增强了用户身份验证的功能和灵活性。

配置`AUTHENTICATION_BACKENDS

为了使用我们的自定义身份验证服务，我们需要配置AUTHENTICATION_BACKENDS，如下所示：

AUTHENTICATION_BACKENDS = (
    'oauth.authentication.EmailOrUsernameModelBackend',
)

创建视图

登录表单

#!/usr/bin/python
# -*- coding: utf-8 -*-
from django.contrib.auth.forms import AuthenticationForm
from django.forms import widgets


class LoginForm(AuthenticationForm):
    """登录表单"""
    def __init__(self, *args, **kwargs):
        super(LoginForm, self).__init__(*args, **kwargs)
        self.fields['username'].widget = widgets.TextInput(attrs={
            'id': 'username',
            'class': 'form-control',
            'aria-errormessage': 'usernameError',
            'placeholder': '用户名'
        })
        self.fields['password'].widget = widgets.PasswordInput(attrs={
            'id': 'password',
            'class': 'form-control',
            'aria-errormessage': 'passwordError',
            'placeholder': '密码'
        })

登录视图

class LoginView(FormView):
    """登录视图"""
    form_class = LoginForm
    template_name = 'oauth/login.html'

    @method_decorator(sensitive_post_parameters('password'))
    @method_decorator(csrf_protect)
    @method_decorator(never_cache)
    def dispatch(self, request, *args, **kwargs):
        return super(LoginView, self).dispatch(request, *args, **kwargs)

    def get_context_data(self, **kwargs):
        kwargs['redirect_to'] = get_redirect_uri(self.request)
        return super(LoginView, self).get_context_data(**kwargs)

    def post(self, request, *args, **kwargs):
        form = LoginForm(data=self.request.POST, request=self.request)
        if form.is_valid():
            auth.login(self.request, form.get_user())
            return super(LoginView, self).form_valid(form)
        return self.render_to_response({
            'form': form
        })

    def get_success_url(self):
        authorize_uri = reverse('authorize', request=self.request, kwargs={
            'authorize_type': 'account'
        })
        data = parse.urlencode({
            'response_type': 'token',
            'redirect_uri': get_redirect_uri(self.request)
        })
        return f'{authorize_uri}?{data}'

注销视图

class LogoutView(RedirectView):
    """退出登录"""

    @method_decorator(never_cache)
    def dispatch(self, request, *args, **kwargs):
        return super(LogoutView, self).dispatch(request, *args, **kwargs)

    def get(self, request, *args, **kwargs):
        user = request.user
        user_token_cache_key = f'oauth:user:id:{user.id}:token'
        if cache.ttl(user_token_cache_key) != 0:
            token = cache.get(user_token_cache_key)
            cache.delete(user_token_cache_key)
            token_user_cache_key = f'oauth:token:{token}:user:id'
            if cache.ttl(token_user_cache_key) != 0:
                cache.delete(token_user_cache_key)
        logout(request)
        return super(LogoutView, self).get(request, *args, **kwargs)

    def get_redirect_url(self, *args, **kwargs):
        return get_redirect_uri(self.request)

用户授权视图

class AuthorizeView(RedirectView):
    """用户授权"""

    @method_decorator(never_cache)
    def dispatch(self, request, *args, **kwargs):
        return super(AuthorizeView, self).dispatch(request, *args, **kwargs)

    def get_redirect_url(self, authorize_type, *args, **kwargs):
        request = self.request
        user = request.user
        token = None
        if authorize_type == 'account':
            if user.is_authenticated:
                token = generate_token()
                token_user_cache_key = f'oauth:token:{token}:user:id'
                user_token_cache_key = f'oauth:user:id:{user.id}:token'
                cache.set(token_user_cache_key, user.id, timeout=60 * 60 * 24)
                cache.set(user_token_cache_key, token, timeout=None)
        if token:
            data = parse.urlencode({
                'access_token': token,
                'token_type': 'bearer'
            })
            return f'{get_redirect_uri(request)}#{data}'
        return reverse('login', request=request)

我们创建了登录表单、登录视图、注销视图和用户授权视图。这些视图提供了用户身份验证和授权的功能。登录表单具有定制的字段，登录视图支持密码和邮箱验证码校验，而注销视图用于用户退出登录。用户授权视图负责用户身份验证后的授权操作。

API视图获取当前登录用户信息

定义用户模型序列化器

#!/usr/bin/python
# -*- coding: utf-8 -*-
from django.contrib.auth import get_user_model
from rest_framework import serializers

User = get_user_model()


class UserSerializer(serializers.ModelSerializer):
    """用户模型列化"""

    class Meta:
        model = User
        exclude = ('password',)
        read_only_fields = ('avatar', 'last_login', 'last_login_ip', 'active')

定义视图

from django.shortcuts import render
from rest_framework import permissions
from rest_framework.generics import GenericAPIView

from CodeVoyager import mixins
from .serializers import UserSerializer


# Create your views here.
class UserProfileAPIView(mixins.UpdateModelMixin, mixins.RetrieveModelMixin, GenericAPIView):
    """用户信息api视图"""
    permission_classes = (permissions.IsAuthenticated,)
    serializer_class = UserSerializer

    def get_object(self):
        user = self.request.user
        if user.is_anonymous or not user.is_authenticated:
            return None
        return self.request.user

    def get(self, request, *args, **kwargs):
        return self.retrieve(request, *args, **kwargs)

    def put(self, request, *args, **kwargs):
        return self.update(request, *args, **kwargs)

    def patch(self, request, *args, **kwargs):
        return self.partial_update(request, *args, **kwargs)