Google App Engine Cloud Tasks and Django Rest Framework permissions

If you use Google App Engine, you can run tasks asynchronously using Cloud Tasks. It’s simple. You can add a task to a queue with some payload which will be send to a specified API endpoint.

It’s essential to secure the API endpoint. It should be accessible only from the Cloud Tasks service. According to the documentation, you can verify the HTTP header.

All requests from Cloud Tasks Service contains X-Appengine-Queuename: true header. If someone tries to send a request with this header to the API, App Engine will remove it.

The X- headers are stripped by App Engine when they originate from external sources so that you can trust this header.

These headers are set internally. If any of these headers are in present in an external user request to your app, they are replaced by the internal ones — except for requests from logged in administrators of the application, who are allowed to set headers for testing purposes.

Using this information, you can create custom DRF permission.

from rest_framework import permissions


class IsTask(permissions.BasePermission):
    def has_permission(self, request, view):
        if request.META.get('HTTP_X_APPENGINE_QUEUENAME'):
            return True
        return False

Use permission_classes decorator to set the IsTask permission.

from rest_framework.decorators import api_view, permission_classes

from core.permissions.task import IsTask


@api_view(["GET"])
@permission_classes([IsTask])
def update_images(request):
    # ...

Now we are sure that the API endpoint can only be used by the Cloud Tasks service.