Django - Verify AWS Cognito token is valid before processing endpoint request

Viewed 631

So I have this code below for checking a AWS Cognito token. I obviously don't want to add these 6 lines of code to every endpoint. Also I don't know if this is the proper way of validating all I'm doing is expecting the token to be of format ' ', parsing it and just decoding the JWT token part. How can I authenticate the AWS amplify token that comes with every request to ensure the user is properly logged in. I'd like to add this authentication to APIView endpoints and DRF api_view decorated endpoints.

views.py

import django.db.utils
from rest_framework import authentication, permissions, status
from rest_framework.views import APIView
from .serializers import *
from .models import *
from rest_framework.response import Response
from django.http import JsonResponse
from drf_yasg.utils import swagger_auto_schema
from drf_yasg import openapi
from .core.api import jwt
from django.core.exceptions import ObjectDoesNotExist
class LoginView(APIView):
    def post(self, request):
        # 'Bearer z324weroko2iorjqoi=+3r3+3ij.2o2ij4='
        token = request.META['HTTP_AUTHORIZATION'].split(' ')[1]
        print(token)
    
        # TODO this should be separated out to a login module
        try:
            res = jwt.decode_cognito_jwt(token)
            return Response(status=status.Http_200_OK)
        except:
            return Response("Invalid JWT", status=status.HTTP_401_UNAUTHORIZED)

@api_view(['GET'])
@swagger_auto_schema(
    operation_description="Get Goals joined by User"
)
def get_goals_by_user(request, user_id):
    try:
        # Get goal ids of user with id
        goals_query = JoinGoal.objects.filter(
            joiner_id=user_id).values_list('goal_id', flat=True)
        goals_list = list(goals_query)
        # Get Goals using list of goals PK with descriptions and uuid
        data = list(Goal.objects.filter(
            pk__in=goals_list).values('description', 'uuid'))
        response_data = dict(goals=data)
        return JsonResponse(response_data, status=status.HTTP_200_OK)
    except JoinGoal.DoesNotExist:
        return Response(dict(error=does_not_exist_msg(JoinGoal.__name__, 'joiner_id', user_id)), status=status.HTTP_400_BAD_REQUEST)
2 Answers

If using djangorestframework, the answer from @bdbd would be your best option. Otherwise, you might want to explore the following options:

  1. Implement your own decorator that will perform the authentication. This has the same idea as the @login_required decorator or the @user_passes_test decorator. When writing such decorator for class-based views, you maybe interested with django.utils.decorators.method_decorator.
from functools import partial, wraps

from django.utils.decorators import method_decorator


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.
        token = request.META['HTTP_AUTHORIZATION'].split(' ')[1]

        try:
            res = jwt.decode_cognito_jwt(token)
            # Authenticate res if valid. Raise exception if not.
        except Exception:
            # Fail if invalid
            return HttpResponseForbidden("You are forbidden here!")
        else:
            # Proceed with the view if valid
            return view_func(request, *args, **kwargs)

    return wrapped_view


# We can decorate it here before the class definition but can also be done before the class method itself. See https://docs.djangoproject.com/en/3.2/topics/class-based-views/intro/#decorating-the-class
@method_decorator(
    name="post",
    decorator=[
        cognito_authenticator,
    ],
)
class SomeView(View):
    @method_decorator(cognito_authenticator)  # As explained above, this is another way of putting the decorator
    def get(self, request):
        return HttpResponse("Allowed entry!")

    def post(self, request):
        return HttpResponse("Allowed entry!")


# Or if using function-based views
@api_view(['POST'])
@cognito_authenticator
def some_view(request):
    return HttpResponse(f"Allowed entry!")
  1. Write a custom middleware. Be aware that the order matters. Same idea as the default AuthenticationMiddleware which populates the request.user field. In your case, implement the __call__ method where you would check the Cognito tokens. Do not proceed to the view when the token is invalid by returning e.g. HttpResponseForbidden as in this reference.
class CognitoAuthenticatorMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        token = request.META['HTTP_AUTHORIZATION'].split(' ')[1]

        try:
            res = jwt.decode_cognito_jwt(token)
            # Authenticate res if valid. Raise exception if not.
        except Exception:
            # Fail if invalid
            return HttpResponseForbidden("You are forbidden here!")

        # Proceed if valid
        response = self.get_response(request)

        return response
MIDDLEWARE = [
    ...
    'path.to.CognitoAuthenticatorMiddleware',
    ...
]

Update

Here is a sample run using Option-1. For simplicity, settings.py is just the default settings.

views.py

from functools import partial, wraps

from django.http import HttpResponse, HttpResponseForbidden
from django.utils.decorators import method_decorator
from django.views import View  # If using django views
from rest_framework.views import APIView  # If using djangorestframework views


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

    @wraps(view_func)
    def wrapped_view(request, *args, **kwargs):
        # To simplify the authentication, we would check if there is a query parameter "name=me". If none, it is forbidden.
        if request.GET.get('name') == "me":
            return view_func(request, *args, **kwargs)
        return HttpResponseForbidden("You are forbidden here!")

    return wrapped_view


@method_decorator(  # Try this style-1
    name="get",
    decorator=[
        cognito_authenticator,
    ],
)
class SomeView(View):  # If using djangorestframework view, this can also inherit from APIView or others e.g. class SomeView(APIView):
    @method_decorator(cognito_authenticator)  # Or try this style-2
    def get(self, request):
        return HttpResponse(f"Allowed entry!")

urls.py

from django.urls import path

from my_app import views

urlpatterns = [
    path("some-view/", views.SomeView.as_view()),
]

Sample run:

$ curl http://127.0.0.1:8000/my_app/some-view/?name=notme
You are forbidden here!
$ curl http://127.0.0.1:8000/my_app/some-view/?name=me
Allowed entry!

Since it seems you are using DRF, you can create your own authentication class and apply the processing of the JWT there:

from django.contrib.auth.models import AnonymousUser
from rest_framework.authentication import BaseAuthentication
from rest_framework import exceptions


class MyCustomJWTAuthentication(BaseAuthentication):
    def authenticate(self, request):
        token = request.META['HTTP_AUTHORIZATION'].split(' ')[1]
        try:
            jwt.decode_cognito_jwt(token)
        except Exception:
            raise exceptions.AuthenticationFailed('Invalid JWT')

        return AnonymousUser(), None


class MyCustomAPIView(APIView):
    authentication_classes = (MyCustomJWTAuthentication, )

Or if you want to apply it to all APIViews:

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': (
        'path.to.MyCustomJWTAuthentication',
    ),
}

Note that once the JWT decode fails, no other authentication classes will be checked. If you don't want this, change the handling for the except clause to not raise AuthenticationFailed.

Related