python - Django - 除非另有说明，否则自动将身份验证器添加到每个端点

Question

所以我创建了这个自定义身份验证器，并且我有超过 30 个端点。对于除 3 个端点之外的所有端点，它都需要身份验证。所以我几乎添加@custom_authenticator到每个函数或类@method_decorator(custom_authenticator)的情况下APIView。有没有办法可以自动将其添加到端点并添加一个装饰器来关闭特定端点功能的身份验证？例如

@donotauth
def endpoint(request)

然后endpoint()不会先运行身份验证器。理想情况下，该解决方案应与下面的自定义身份验证器一起使用

自定义身份验证器

def cognito_authenticator(view_func=None):
    if view_func is None:
        return partial(cognito_authenticator)

    @wraps(view_func)
    def wrapped_view(request, *args, **kwargs):
        # Check the cognito token from the request.
        auth = request.headers.get("Authorization", None)
        if not auth:
            return Response(dict(error='Authorization header expected'), status=status.HTTP_401_UNAUTHORIZED)

        parts = auth.split()

        if parts[0].lower() != "bearer":
            return Response(dict(error='Authorization header must start with bearer'),
                            status=status.HTTP_401_UNAUTHORIZED)
        elif len(parts) == 1:
            return Response(dict(error='Token not found'), status=status.HTTP_401_UNAUTHORIZED)
        elif len(parts) > 2:
            return Response(dict(error='Authorization header must be Bearer token'),
                            status=status.HTTP_401_UNAUTHORIZED)

        token = parts[1]
        try:
            res = decode_cognito_jwt(token)
            expiration = datetime.utcfromtimestamp(res['exp'])
            current_utc = datetime.utcnow()

            if current_utc > expiration:
                return Response(dict(error=f'current time:{current_utc} is after expiration:{expiration}',
                                     user_msg='Please login again'), status=status.HTTP_400_BAD_REQUEST)

        except Exception:
            # Fail if invalid
            return Response(dict(error="Invalid JWT"),
                            status=status.HTTP_401_UNAUTHORIZED)  # Or HttpResponseForbidden()
        else:
            # Proceed with the view if valid
            return view_func(request, *args, **kwargs)

    return wrapped_view

使用中间件的解决方案 1：

我尝试添加中间件，但它会引发任何带有@api_view装饰器的错误。我得到的错误是AssertionError: .accepted_renderer not set on Response.如何在每个端点上设置我的自定义身份验证，无论它是否具有@api_view装饰器或APIView. 最终目标应该是自动将上述内容添加cognito_authenticator到任何端点，并指定何时不使用身份验证器（可能是功能性装饰器）

视图.py

@api_view(['GET'])
@swagger_auto_schema(
    operation_description="Get <count> most recent posts by category"
)
def get_most_recent_posts_by_category(request, category, count):
    return Response(status=status.HTTP_200_OK)

中间件

from datetime import datetime

from rest_framework import status
from rest_framework.response import Response

from cheers.core.api.jwt_helpers import decode_cognito_jwt


class CognitoMiddleware(object):
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        return self.get_response(request)

    def process_view(self, request, view_func, view_args, view_kwargs):
        auth = request.headers.get("Authorization", None)
        if not auth:
            return Response(dict(error='Authorization header expected'), status=status.HTTP_401_UNAUTHORIZED)

        parts = auth.split()

        if parts[0].lower() != "bearer":
            return Response(dict(error='Authorization header must start with bearer'),
                            status=status.HTTP_401_UNAUTHORIZED)
        elif len(parts) == 1:
            return Response(dict(error='Token not found'), status=status.HTTP_401_UNAUTHORIZED)
        elif len(parts) > 2:
            return Response(dict(error='Authorization header must be Bearer token'),
                            status=status.HTTP_401_UNAUTHORIZED)

        token = parts[1]
        try:
            res = decode_cognito_jwt(token)
            expiration = datetime.utcfromtimestamp(res['exp'])
            current_utc = datetime.utcnow()

            if current_utc > expiration:
                return Response(dict(error=f'current time:{current_utc} is after expiration:{expiration}',
                                     user_msg='Please login again'), status=status.HTTP_400_BAD_REQUEST)

        except Exception:
            # Fail if invalid
            return Response(dict(error="Invalid JWT"),
                            status=status.HTTP_401_UNAUTHORIZED)  # Or HttpResponseForbidden()
        else:
            # Proceed with the view if valid
            return None

设置.py

MIDDLEWARE = [
    'django.middleware.common.CommonMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'cheers.middleware.CognitoMiddleware.CognitoMiddleware'
]

解决方案 2 使用身份验证器

验证器.py

class CognitoAuthentication(BaseAuthentication):
    def authenticate(self, request):
        auth = request.headers.get("Authorization", None)
        if not auth:
            return Response(dict(error='Authorization header expected'), status=status.HTTP_401_UNAUTHORIZED)

        parts = auth.split()

        if parts[0].lower() != "bearer":
            return Response(dict(error='Authorization header must start with bearer'),
                            status=status.HTTP_401_UNAUTHORIZED)
        elif len(parts) == 1:
            return Response(dict(error='Token not found'), status=status.HTTP_401_UNAUTHORIZED)
        elif len(parts) > 2:
            return Response(dict(error='Authorization header must be Bearer token'),
                            status=status.HTTP_401_UNAUTHORIZED)

        token = parts[1]
        try:
            res = decode_cognito_jwt(token)
            expiration = datetime.utcfromtimestamp(res['exp'])
            current_utc = datetime.utcnow()

            if current_utc > expiration:
                return Response(dict(error=f'current time:{current_utc} is after expiration:{expiration}',
                                     user_msg='Please login again'), status=status.HTTP_400_BAD_REQUEST)

        except Exception:
            # Fail if invalid
            return Response(dict(error="Invalid JWT"),
                            status=status.HTTP_401_UNAUTHORIZED)  # Or HttpResponseForbidden()
        else:
            # Proceed with the view if valid
            return AnonymousUser(), None

设置.py

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': (
        'cheers.utils.authenticator.CognitoAuthentication',
    ),
}

但是在APIView发布功能上它给出了错误django.template.response.ContentNotRenderedError: The response content must be rendered before it can be iterated over.

标签： pythondjangodjango-rest-frameworkdjango-views

Answer 1

要在 DRF API 端点中添加不同级别的身份验证，您可以使用项目级别身份验证和视图级别身份验证：

# config/settings.py
REST_FRAMEWORK = {
'DEFAULT_PERMISSION_CLASSES': [
    # 'rest_framework.permissions.AllowAny', to allow all
    # 'rest_framework.permissions.IsAdminUser', only admin
    # 'rest_framework.permissions.IsAuthenticatedOrReadOnly', only authenticated can write
    'rest_framework.permissions.IsAuthenticated',
    ]
}

# app_name/views.py
@permission_classes((IsAdminUser, ))
def example_view(request, format=None):
    content = {
        'status': 'request was permitted'
    }
    return Response(content)

在这种情况下，所有端点都受IsAuthenticated权限保护。但是视图example_view会覆盖全局权限并IsAdminUser改为使用。