feat: add AuthZ permissions to course creation and outline

cms/djangoapps/contentstore/api/tests/test_validation.py

@@ -11,15 +11,15 @@
 from django.contrib.auth import get_user_model
 from django.test.utils import override_settings
 from django.urls import reverse
-from openedx_authz.constants.roles import COURSE_DATA_RESEARCHER, COURSE_STAFF
+from openedx_authz.constants.roles import COURSE_DATA_RESEARCHER, COURSE_EDITOR, COURSE_STAFF
 from rest_framework import status
 from rest_framework.test import APIClient, APITestCase
 
 from cms.djangoapps.contentstore.api.tests.base import BaseCourseViewTest
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.course_modes.tests.factories import CourseModeFactory
 from common.djangoapps.student.tests.factories import StaffFactory, UserFactory
-from openedx.core.djangoapps.authz.tests.mixins import CourseAuthzTestMixin
+from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin, CourseAuthzTestMixin
 from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
 from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
 
@@ -247,7 +247,7 @@ def test_create_update_reference_success(self, mock_block, mock_user_task_status
 
         mock_auth.assert_called_once()
 
-    @patch('cms.djangoapps.contentstore.api.views.utils.has_course_author_access')
+    @patch('openedx.core.djangoapps.authz.decorators.user_has_course_permission')
     @patch('xmodule.library_content_block.LegacyLibraryContentBlock.is_ready_to_migrate_to_v2')
     def test_list_ready_to_update_reference_success(self, mock_block, mock_auth):
         """
@@ -353,3 +353,83 @@ def test_non_staff_user_cannot_access(self):
 
         resp = non_staff_client.get(self.get_url(self.course_key))
         self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)  # noqa: PT009
+
+
+class TestMigrationViewSetCreateAuthz(
+    CourseAuthoringAuthzTestMixin,
+    SharedModuleStoreTestCase,
+    APITestCase,
+):
+    """
+    AuthZ tests for:
+    /api/courses/v1/migrate_legacy_content_blocks/<course_id>/
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+
+        cls.course = CourseFactory.create(
+            display_name='test course',
+            run="Testing_course",
+        )
+        cls.course_key = cls.course.id
+
+        cls.initialize_course(cls.course)
+
+    @classmethod
+    def initialize_course(cls, course):
+        """Sets up test course structure."""
+        section = BlockFactory.create(
+            parent_location=course.location,
+            category="chapter",
+        )
+        subsection = BlockFactory.create(
+            parent_location=section.location,
+            category="sequential",
+        )
+        unit = BlockFactory.create(
+            parent_location=subsection.location,
+            category="vertical",
+        )
+        BlockFactory.create(
+            parent_location=unit.location,
+            category="library_content",
+        )
+
+    def url(self):
+        return f"/api/courses/v1/migrate_legacy_content_blocks/{self.course_key}/"
+
+    # ---- GET (list) ----
+
+    def test_authorized_user_can_list_blocks(self):
+        """Authorized user can list migratable blocks."""
+        self.add_user_to_role_in_course(
+            self.authorized_user,
+            COURSE_EDITOR.external_key,
+            self.course.id,
+        )
+
+        response = self.authorized_client.get(self.url())
+
+        assert response.status_code == status.HTTP_200_OK
+
+    def test_unauthorized_user_cannot_list_blocks(self):
+        """Unauthorized user should receive 403."""
+        response = self.unauthorized_client.get(self.url())
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    # ---- elevated users ----
+
+    def test_staff_user_can_access_without_authz_role(self):
+        """Staff user bypasses AuthZ."""
+        response = self.staff_client.get(self.url())
+
+        assert response.status_code == status.HTTP_200_OK
+
+    def test_superuser_can_access_without_authz_role(self):
+        """Superuser bypasses AuthZ."""
+        response = self.super_client.get(self.url())
+
+        assert response.status_code in [status.HTTP_200_OK, status.HTTP_201_CREATED]

cms/djangoapps/contentstore/api/views/course_validation.py

@@ -364,7 +364,7 @@ class CourseLegacyLibraryContentSerializer(serializers.Serializer):
     usage_key = serializers.CharField()
 
 
-class CourseLegacyLibraryContentMigratorView(StatusViewSet):
+class CourseLegacyLibraryContentMigratorView(DeveloperErrorViewMixin, StatusViewSet):
     """
     This endpoint is used for migrating legacy library content to the new item bank block library v2.
     """
@@ -384,7 +384,7 @@ class CourseLegacyLibraryContentMigratorView(StatusViewSet):
             401: "The requester is not authenticated.",
         },
     )
-    @course_author_access_required
+    @authz_permission_required(COURSES_VIEW_COURSE.identifier, LegacyAuthoringPermission.WRITE)
     def list(self, _, course_key):  # pylint: disable=arguments-differ
         """
         Returns all legacy library content blocks ready to be migrated to new item bank block.

cms/djangoapps/contentstore/rest_api/v1/views/course_index.py

@@ -5,6 +5,7 @@
 import edx_api_doc_tools as apidocs
 from django.conf import settings
 from opaque_keys.edx.keys import CourseKey
+from openedx_authz.constants.permissions import COURSES_VIEW_COURSE
 from rest_framework.fields import BooleanField
 from rest_framework.request import Request
 from rest_framework.response import Response
@@ -22,7 +23,7 @@
 )
 from cms.djangoapps.contentstore.xblock_storage_handlers.view_handlers import get_xblock
 from cms.lib.xblock.upstream_sync import UpstreamLink
-from common.djangoapps.student.auth import has_studio_read_access
+from openedx.core.djangoapps.authz.decorators import LegacyAuthoringPermission, user_has_course_permission
 from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin, verify_course_exists, view_auth_classes
 from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.exceptions import ItemNotFoundError  # lint-amnesty, pylint: disable=wrong-import-order
@@ -101,7 +102,12 @@ def get(self, request: Request, course_id: str):
         """
 
         course_key = CourseKey.from_string(course_id)
-        if not has_studio_read_access(request.user, course_key):
+        if not user_has_course_permission(
+            request.user,
+            COURSES_VIEW_COURSE.identifier,
+            course_key,
+            LegacyAuthoringPermission.READ
+        ):
             self.permission_denied(request)
         course_index_context = get_course_index_context(request, course_key)
         course_index_context.update({

cms/djangoapps/contentstore/rest_api/v1/views/tests/test_course_index.py

@@ -5,6 +5,7 @@
 from django.test import RequestFactory
 from django.urls import reverse
 from edx_toggles.toggles.testutils import override_waffle_flag
+from openedx_authz.constants.roles import COURSE_EDITOR
 from rest_framework import status
 
 from cms.djangoapps.contentstore.config.waffle import CUSTOM_RELATIVE_DATES
@@ -13,6 +14,7 @@
 from cms.djangoapps.contentstore.utils import get_lms_link_for_item, get_pages_and_resources_url
 from cms.djangoapps.contentstore.views.course import _course_outline_json
 from common.djangoapps.student.tests.factories import UserFactory
+from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin
 from openedx.core.djangoapps.waffle_utils.testutils import WAFFLE_TABLES
 from xmodule.modulestore.tests.factories import BlockFactory, check_mongo_calls
 
@@ -162,3 +164,63 @@ def test_number_of_calls_to_db(self):
         with self.assertNumQueries(34, table_ignorelist=WAFFLE_TABLES):
             with check_mongo_calls(3):
                 self.client.get(self.url)
+
+
+class CourseIndexAuthzViewTest(CourseAuthoringAuthzTestMixin, CourseTestCase):
+    """
+    Tests for CourseIndexView using AuthZ permissions.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.url = reverse(
+            "cms.djangoapps.contentstore:v1:course_index",
+            kwargs={"course_id": self.course.id},
+        )
+
+    def test_authorized_user_can_access_course_index(self):
+        """Authorized user with COURSE_EDITOR role can access course index."""
+        self.add_user_to_role_in_course(
+            self.authorized_user,
+            COURSE_EDITOR.external_key,
+            self.course.id
+        )
+
+        response = self.authorized_client.get(self.url)
+
+        assert response.status_code == status.HTTP_200_OK
+        assert "course_structure" in response.data
+
+    def test_unauthorized_user_cannot_access_course_index(self):
+        """Unauthorized user should receive 403."""
+        response = self.unauthorized_client.get(self.url)
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_user_without_role_then_added_can_access(self):
+        """Validate dynamic role assignment works as expected."""
+        response = self.unauthorized_client.get(self.url)
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+        self.add_user_to_role_in_course(
+            self.unauthorized_user,
+            COURSE_EDITOR.external_key,
+            self.course.id
+        )
+
+        response = self.unauthorized_client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+
+    def test_staff_user_can_access_without_authz_role(self):
+        """Django staff user should access without AuthZ role."""
+        response = self.staff_client.get(self.url)
+
+        assert response.status_code == status.HTTP_200_OK
+        assert "course_structure" in response.data
+
+    def test_superuser_can_access_without_authz_role(self):
+        """Superuser should access without AuthZ role."""
+        response = self.super_client.get(self.url)
+
+        assert response.status_code == status.HTTP_200_OK
+        assert "course_structure" in response.data

cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py

@@ -90,6 +90,7 @@
 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 rest_framework.exceptions import NotFound, PermissionDenied, ValidationError
 from rest_framework.fields import BooleanField
 from rest_framework.request import Request
@@ -115,6 +116,7 @@
 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
 from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin, view_auth_classes
@@ -302,7 +304,12 @@ def get(self, request: _AuthenticatedRequest, course_key_string: str):
         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_COURSE.identifier,
+            course_key,
+            LegacyAuthoringPermission.READ
+        ):
             raise PermissionDenied
 
         # Gets all links of the Course, using the

cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py

@@ -11,8 +11,10 @@
 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_content import models_api as content_models
 from organizations.models import Organization
+from rest_framework import status
 
 from cms.djangoapps.contentstore.helpers import StaticFileNotices
 from cms.djangoapps.contentstore.tests.utils import CourseTestCase
@@ -22,6 +24,7 @@
 from common.djangoapps.student.auth import add_users
 from common.djangoapps.student.roles import CourseStaffRole
 from common.djangoapps.student.tests.factories import UserFactory
+from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin
 from openedx.core.djangoapps.content_libraries import api as lib_api
 from xmodule.modulestore.django import modulestore
 from xmodule.modulestore.tests.django_utils import ImmediateOnCommitMixin, SharedModuleStoreTestCase
@@ -1517,6 +1520,68 @@ def test_200_summary(self):
         self.assertListEqual(data, expected)  # noqa: PT009
 
 
+class GetDownstreamSummaryAuthzViewTest(
+    CourseAuthoringAuthzTestMixin,
+    _BaseDownstreamViewTestMixin,
+    ImmediateOnCommitMixin,
+    SharedModuleStoreTestCase,
+):
+    """
+    AuthZ tests for:
+    GET /api/contentstore/v2/downstreams/<course_id>/summary
+    """
+
+    def call_api(self, client, course_id): # pylint: disable=arguments-differ
+        return client.get(f"/api/contentstore/v2/downstreams/{course_id}/summary")
+
+    def test_authorized_user_can_access_summary(self):
+        """Authorized user with COURSE_EDITOR role can access summary."""
+        self.add_user_to_role_in_course(
+            self.authorized_user,
+            COURSE_EDITOR.external_key,
+            self.course.id
+        )
+
+        response = self.call_api(self.authorized_client, str(self.course.id))
+
+        assert response.status_code == status.HTTP_200_OK
+        assert isinstance(response.json(), list)
+
+    def test_unauthorized_user_cannot_access_summary(self):
+        """Unauthorized user should receive 403."""
+        response = self.call_api(self.unauthorized_client, str(self.course.id))
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_user_without_role_then_added_can_access(self):
+        """Validate dynamic role assignment works."""
+        response = self.call_api(self.unauthorized_client, str(self.course.id))
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+        self.add_user_to_role_in_course(
+            self.unauthorized_user,
+            COURSE_EDITOR.external_key,
+            self.course.id
+        )
+
+        response = self.call_api(self.unauthorized_client, str(self.course.id))
+        assert response.status_code == status.HTTP_200_OK
+
+    def test_staff_user_can_access_without_authz_role(self):
+        """Staff user should access without explicit AuthZ role."""
+        response = self.call_api(self.staff_client, str(self.course.id))
+
+        assert response.status_code == status.HTTP_200_OK
+        assert isinstance(response.json(), list)
+
+    def test_superuser_can_access_without_authz_role(self):
+        """Superuser should access without explicit AuthZ role."""
+        response = self.call_api(self.super_client, str(self.course.id))
+
+        assert response.status_code == status.HTTP_200_OK
+        assert isinstance(response.json(), list)
+
+
 class GetDownstreamDeletedUpstream(
     _BaseDownstreamViewTestMixin,
     ImmediateOnCommitMixin,