Skip to content

Commit

Permalink
Merge pull request #848 from openedx/asheehan-edx/get-query-by-hash-p…
Browse files Browse the repository at this point in the history
…ermissions-cleanup

feat: fixing top level permission validation from the get query by hash endpoint
alex-sheehan-edx authored Jun 14, 2024
2 parents bd9a58f + b6d6a1a commit 1d4801c
Showing 2 changed files with 62 additions and 5 deletions.
17 changes: 15 additions & 2 deletions enterprise_catalog/apps/api/v1/tests/test_views.py
Original file line number Diff line number Diff line change
@@ -2512,7 +2512,7 @@ def setUp(self):
self.set_up_catalog_learner()
self.catalog_query_object = CatalogQueryFactory()
self.catalog_object = EnterpriseCatalogFactory(catalog_query=self.catalog_query_object)
self.assign_catalog_admin_feature_role(enterprise_uuids=[self.catalog_object.enterprise_uuid])
self.assign_catalog_admin_jwt_role(str(self.catalog_object.enterprise_uuid))
# Factory doesn't set up a hash, so do it manually
self.catalog_query_object.content_filter_hash = get_content_filter_hash(
self.catalog_query_object.content_filter
@@ -2556,6 +2556,10 @@ def test_get_query_by_hash(self):
response = self.client.get(url)
assert response.status_code == 404

self.client.logout()
response = self.client.get(url)
assert response.status_code == 401

def test_get_query_by_hash_not_found(self):
"""
Test that the get query by hash endpoint returns expected not found
@@ -2611,6 +2615,10 @@ def test_catalog_query_retrieve(self):
response_json = response.json()
assert response_json.get('uuid') == str(different_customer_catalog.catalog_query.uuid)

self.client.logout()
response = self.client.get(url)
assert response.status_code == 401

def test_catalog_query_list(self):
"""
Test that the Catalog Query viewset supports listing queries
@@ -2620,6 +2628,7 @@ def test_catalog_query_list(self):
self.assign_catalog_admin_jwt_role(
self.enterprise_uuid,
self.catalog_query_object.enterprise_catalogs.first().enterprise_uuid,
self.catalog_object.enterprise_uuid,
)
url = reverse('api:v1:catalog-queries-list')
response = self.client.get(url)
@@ -2636,4 +2645,8 @@ def test_catalog_query_list(self):
self.set_up_invalid_jwt_role()
self.remove_role_assignments()
response = self.client.get(url)
assert response.data == {'count': 0, 'next': None, 'previous': None, 'results': []}
assert response.json() == {'count': 0, 'next': None, 'previous': None, 'results': []}

self.client.logout()
response = self.client.get(url)
assert response.status_code == 401
50 changes: 47 additions & 3 deletions enterprise_catalog/apps/api/v1/views/catalog_query.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from django.utils.decorators import method_decorator
from django.utils.functional import cached_property
from edx_rbac.mixins import PermissionRequiredForListingMixin
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.exceptions import NotFound
@@ -13,17 +14,34 @@
CatalogQueryGetByHashRequestSerializer,
CatalogQuerySerializer,
)
from enterprise_catalog.apps.catalog.models import CatalogQuery
from enterprise_catalog.apps.api.v1.views.base import BaseViewSet
from enterprise_catalog.apps.catalog.models import (
CatalogQuery,
EnterpriseCatalog,
EnterpriseCatalogRoleAssignment,
)
from enterprise_catalog.apps.catalog.rules import (
enterprises_with_admin_access,
has_access_to_all_enterprises,
)


class CatalogQueryViewSet(viewsets.ReadOnlyModelViewSet):
class CatalogQueryViewSet(viewsets.ReadOnlyModelViewSet, BaseViewSet, PermissionRequiredForListingMixin):
"""Read-only viewset for Catalog Query records"""
renderer_classes = [JSONRenderer]
serializer_class = CatalogQuerySerializer
permission_required = 'catalog.has_admin_access'
list_lookup_field = 'enterprise_catalogs__enterprise_uuid'
role_assignment_class = EnterpriseCatalogRoleAssignment

@property
def base_queryset(self):
"""
Required by the `PermissionRequiredForListingMixin`.
For non-list actions, this is what's returned by `get_queryset()`.
For list actions, some non-strict subset of this is what's returned by `get_queryset()`.
"""
return CatalogQuery.objects.all()

@cached_property
def admin_accessible_enterprises(self):
@@ -37,7 +55,7 @@ def get_queryset(self):
Restrict the queryset to catalog queries the requesting user has access to. Iff the user is staff they have
access to all queries.
"""
all_queries = CatalogQuery.objects.all()
all_queries = self.base_queryset
if not self.admin_accessible_enterprises:
return CatalogQuery.objects.none()
if has_access_to_all_enterprises(self.admin_accessible_enterprises) or self.request.user.is_staff:
@@ -46,6 +64,32 @@ def get_queryset(self):
enterprise_catalogs__enterprise_uuid__in=self.admin_accessible_enterprises
)

def check_permissions(self, request):
"""
If dealing with a "list" action, goes through some customized
logic to check which contexts are accessible by the requesting
user. If none are, and the user is not staff/super, raise
a `PermissionDenied` exception. Uses the parent class's `check_permissions()`
method if the request action is not "list".
"""
if request.user.is_staff and self.staff_are_never_forbidden:
return
if request.user.is_superuser:
return
if self.request_action == 'list':
if not self.accessible_contexts:
self.permission_denied(request)
else:
related_catalog_uuids = [
str(cat.enterprise_uuid) for cat in EnterpriseCatalog.objects.filter(
catalog_query__in=self.get_queryset()
)
]
allowed_contexts = self.admin_accessible_enterprises
if set(related_catalog_uuids).intersection(allowed_contexts):
return
self.permission_denied(request)

@method_decorator(require_at_least_one_query_parameter('hash'))
@action(detail=True, methods=['get'])
def get_query_by_hash(self, request, **kwargs):

0 comments on commit 1d4801c

Please sign in to comment.