diff --git a/cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py b/cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py index 8a669a782ed6..66cdb67b9192 100644 --- a/cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py +++ b/cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py @@ -90,7 +90,10 @@ from opaque_keys import InvalidKeyError from opaque_keys.edx.keys import CourseKey, UsageKey from opaque_keys.edx.locator import LibraryContainerLocator, LibraryLocatorV2, LibraryUsageLocatorV2 -from openedx_authz.constants.permissions import COURSES_VIEW_COURSE +from openedx_authz.constants.permissions import ( + COURSES_MANAGE_LIBRARY_UPDATES, + COURSES_VIEW_LIBRARY_UPDATES, +) from rest_framework.exceptions import NotFound, PermissionDenied, ValidationError from rest_framework.fields import BooleanField from rest_framework.request import Request @@ -115,7 +118,6 @@ ) from cms.lib.xblock.upstream_sync_block import fetch_customizable_fields_from_block from cms.lib.xblock.upstream_sync_container import fetch_customizable_fields_from_container -from common.djangoapps.student.auth import has_studio_read_access, has_studio_write_access from openedx.core.djangoapps.authz.decorators import LegacyAuthoringPermission, user_has_course_permission from openedx.core.djangoapps.content_libraries import api as lib_api from openedx.core.djangoapps.video_config.transcripts_utils import clear_transcripts @@ -198,7 +200,12 @@ def get(self, request: _AuthenticatedRequest): except InvalidKeyError as exc: raise ValidationError(detail=f"Malformed course key: {course_key_string}") from exc - if not has_studio_read_access(request.user, course_key): + if not user_has_course_permission( + request.user, + COURSES_VIEW_LIBRARY_UPDATES.identifier, + course_key, + LegacyAuthoringPermission.READ + ): raise PermissionDenied if ready_to_sync is not None: link_filter["ready_to_sync"] = BooleanField().to_internal_value(ready_to_sync) @@ -306,7 +313,7 @@ def get(self, request: _AuthenticatedRequest, course_key_string: str): if not user_has_course_permission( request.user, - COURSES_VIEW_COURSE.identifier, + COURSES_VIEW_LIBRARY_UPDATES.identifier, course_key, LegacyAuthoringPermission.READ ): @@ -568,10 +575,22 @@ def _load_accessible_block(user: User, usage_key_string: str, *, require_write_a usage_key = UsageKey.from_string(usage_key_string) except InvalidKeyError as exc: raise ValidationError(detail=f"Malformed block usage key: {usage_key_string}") from exc - if require_write_access and not has_studio_write_access(user, usage_key.context_key): - raise not_found - if not has_studio_read_access(user, usage_key.context_key): + + context_key = usage_key.context_key + if not isinstance(context_key, CourseKey): raise not_found + + if require_write_access: + if not user_has_course_permission( + user, COURSES_MANAGE_LIBRARY_UPDATES.identifier, context_key, LegacyAuthoringPermission.WRITE + ): + raise not_found + else: + if not user_has_course_permission( + user, COURSES_VIEW_LIBRARY_UPDATES.identifier, context_key, LegacyAuthoringPermission.READ + ): + raise not_found + try: block = modulestore().get_item(usage_key) except ItemNotFoundError as exc: diff --git a/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py b/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py index 149732437215..dc609c5c3ba5 100644 --- a/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py +++ b/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py @@ -11,7 +11,7 @@ from freezegun import freeze_time from opaque_keys.edx.keys import ContainerKey, UsageKey from opaque_keys.edx.locator import LibraryLocatorV2, LibraryUsageLocatorV2 -from openedx_authz.constants.roles import COURSE_EDITOR +from openedx_authz.constants.roles import COURSE_AUDITOR, COURSE_EDITOR from openedx_content import models_api as content_models from organizations.models import Organization from rest_framework import status @@ -1686,3 +1686,65 @@ def test_delete_component_should_be_ready_to_sync(self): } self.assertDictEqual(data[0], expected_results) # noqa: PT009 + + +class GetDownstreamListAuthzViewTest( + CourseAuthoringAuthzTestMixin, + _BaseDownstreamViewTestMixin, + ImmediateOnCommitMixin, + SharedModuleStoreTestCase, +): + """ + AuthZ tests for: + GET /api/contentstore/v2/downstreams/?course_id=... + + Validates that view_library_updates grants read access and + manage_library_updates is required for sync operations. + """ + + def call_list_api(self, client, course_id): + return client.get("/api/contentstore/v2/downstreams/", data={"course_id": str(course_id)}) + + def call_sync_api(self, client, usage_key): + return client.post( + f"/api/contentstore/v2/downstreams/{usage_key}/sync", + content_type="application/json", + ) + + def test_editor_can_list_downstreams(self): + """Course editor (has view_library_updates) can list downstream links.""" + self.add_user_to_role_in_course( + self.authorized_user, COURSE_EDITOR.external_key, self.course.id + ) + response = self.call_list_api(self.authorized_client, self.course.id) + assert response.status_code == status.HTTP_200_OK + + def test_auditor_can_list_downstreams(self): + """Course auditor (has view_library_updates) can list downstream links.""" + self.add_user_to_role_in_course( + self.authorized_user, COURSE_AUDITOR.external_key, self.course.id + ) + response = self.call_list_api(self.authorized_client, self.course.id) + assert response.status_code == status.HTTP_200_OK + + def test_unauthorized_user_cannot_list_downstreams(self): + """User without any course role cannot list downstream links.""" + response = self.call_list_api(self.unauthorized_client, self.course.id) + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_editor_can_sync_downstream(self): + """Course editor (has manage_library_updates) can sync a downstream block.""" + self.add_user_to_role_in_course( + self.authorized_user, COURSE_EDITOR.external_key, self.course.id + ) + response = self.call_sync_api(self.authorized_client, str(self.downstream_video_key)) + # 200 = sync success, 400 = validation error (e.g. bad upstream) — both confirm permission was granted + assert response.status_code in (status.HTTP_200_OK, status.HTTP_400_BAD_REQUEST) + + def test_auditor_cannot_sync_downstream(self): + """Course auditor (only view_library_updates) cannot sync a downstream block.""" + self.add_user_to_role_in_course( + self.authorized_user, COURSE_AUDITOR.external_key, self.course.id + ) + response = self.call_sync_api(self.authorized_client, str(self.downstream_video_key)) + assert response.status_code == status.HTTP_404_NOT_FOUND