feat: split read vs write authz checks for library updates

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:

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