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)