Tastypie Django full hydrate returns an error when joining the protected Resource

26 Views Asked by At

I created here a small example to explain my problem for which I haven't find a solution on the web. I am using Django and Tastypie to implement a REST API. I have MyUserResource which is a descendant of Django's User model and this resource is protected with StaffAuthorization. Then I have CommentsResource which should be opened for Anonymous users as well.

Now, my problem lies in the fact that during running Tastypie's full_hydrate operation within obj_create, Tastypie is trying to make a join on user table. That means that in case anonymous users are trying to fetch comments, the API returns Unauthorized error. For now I implement it to "open" MyUserResource by using Authorization and checking within obj_create, obj_update and obj_delete if the user is staff or a superuser. But I wouldn't want that everyone is able to retrieve the list of users as well. Is it possible to protect User resource and still allow joins during hydration process? I still haven't tried to implement "my" obj_get function and there check if users are staff or superusers and for other types of user I might remove sensitive data?

Here is the sample code, first Django models and then Tastypie resources. At the very end I provide a code for my implementation of StaffAuthorization which is a specialization of Authorization class:

class MyUser(AbstractUser):
    location = models.PointField(srid=3857, null=True, blank=True)
    geo_coverage = models.PolygonField(srid=3857, null=True, blank=True)
    address1 = models.CharField(max_length=200, null=True, blank=True)
    address2 = models.CharField(max_length=200, null=True, blank=True)
    region = models.CharField(max_length=200, null=True, blank=True)
    city = models.CharField(max_length=100, null=True, blank=True)
    zip_code = models.CharField(max_length=200, null=True, blank=True)
    country = models.CharField(max_length=200, null=True, blank=True)
    avatar = models.CharField(max_length=200, null=True, blank=True)

    class Meta:
        db_table = 'my_user'

class Comment():
    # This join return an error during full_hydrate
    contributor = models.ForeignKey(MyUser, on_delete=models.CASCADE, null=False, blank=False)
    status = models.ForeignKey(CommentStatus, on_delete=models.SET_DEFAULT, null=False, blank=False, default=1)
    content = models.CharField(max_length=200, null=False, blank=False)


class MyUserResource(Resource):
        
    class Meta:
        list_allowed_methods = ('get', 'post',)
        detail_allowed_methods = ('get', 'post', 'patch', 'delete',)
        excludes = ['password']
        authorization = StaffAuthorization()
        authentication = MultiAuthentication(
            ApiKeyAuthentication(),
            CookieBasicAuthentication(),
        )

        resource_name = 'myuser'
        queryset = MyUser.objects.all()
        always_return_data = True

        filtering = {
            "id": ('exact',),
            "role": ALL_WITH_RELATIONS,
        }

class CommentResource(Resource):
    contributor = fields.ToOneField(
        MyUserResource,
        attribute='contributor',
        full=True,
        null=True
    )
    status = fields.ToOneField(
        CommentStatusResource,
        attribute='status',
        full=True,
        null=True
    )

    class Meta:
        list_allowed_methods = ('get', 'post', 'patch', 'delete',)
        detail_allowed_methods = ('get', 'post', 'patch', 'delete',)
        authorization = Authorization()
        authentication = MultiAuthentication(
            ApiKeyAuthentication(),
            CookieBasicAuthentication(),
            Authentication(),
        )
        resource_name = 'comment'
        queryset = Comment.objects.all()
        always_return_data = True

        filtering = {
            "id": ('exact',),
            "contributor": ALL_WITH_RELATIONS,
        }
        ordering = ['status', 'contributor']


class StaffAuthorization(Authorization):
    """
    Class for staff authorization.
    """

    def authorized(self, object_list, bundle):
        """Checks if a user is superuser of staff member."""
        user = bundle.request.user
        try:
            return user.is_active and (user.is_superuser or user.is_staff)
        except AttributeError:
            raise Unauthorized('You have to authenticate first!')

    def authorized_list_auth(self, object_list, bundle):
        """
        Returns object_list for superusers or staff members,
        otherwise returns empty list.
        """
        if self.authorized(object_list, bundle):
            return object_list
        return []

    def read_list(self, object_list, bundle):
        """
        Returns data from object_list for superusers or staff members.
        This assumes a QuerySet from ModelResource, therefore tries to return
        .all(), or original object_list in case of failure.
        """
        try:
            object_list = object_list.all()
        except AttributeError:
            # dict doesn't have .all()
            pass
        return self.authorized_list_auth(object_list, bundle)

    def read_detail(self, object_list, bundle):
        if bundle.request.user.is_anonymous:
            # Double-check anonymous users, because operations
            # on embedded fields do not pass through authentication.
            ApiKeyAuthentication().is_authenticated(bundle.request)
        return self.authorized(object_list, bundle)

    def create_list(self, object_list, bundle):
        return self.authorized_list_auth(object_list, bundle)

    def create_detail(self, object_list, bundle):
        return self.authorized(object_list, bundle)

    def update_list(self, object_list, bundle):
        return self.authorized_list_auth(object_list, bundle)

    def update_detail(self, object_list, bundle):
        return self.authorized(object_list, bundle)

    def delete_list(self, object_list, bundle):
        """Superuser and staff can delete lists."""
        return self.authorized_list_auth(object_list, bundle)

    def delete_detail(self, object_list, bundle):
        """Superuser and staff can delete item."""
        return self.authorized(object_list, bundle)
0

There are 0 best solutions below