From b657f5822216dbc1accab0ab6291986f0e54f889 Mon Sep 17 00:00:00 2001 From: Daniel Wong Date: Tue, 31 Mar 2026 11:27:44 -0600 Subject: [PATCH] feat: add AuthZ permissions to course creation and outline --- .../contentstore/api/tests/test_validation.py | 89 ++++++++++++++++++- .../api/views/course_validation.py | 4 +- .../rest_api/v1/views/course_index.py | 11 ++- .../v1/views/tests/test_course_index.py | 62 +++++++++++++ .../rest_api/v2/views/downstreams.py | 9 +- .../v2/views/tests/test_downstreams.py | 65 ++++++++++++++ openedx/core/djangoapps/authz/tests/mixins.py | 8 ++ 7 files changed, 240 insertions(+), 8 deletions(-) diff --git a/cms/djangoapps/contentstore/api/tests/test_validation.py b/cms/djangoapps/contentstore/api/tests/test_validation.py index a091eedcc577..31af868e240e 100644 --- a/cms/djangoapps/contentstore/api/tests/test_validation.py +++ b/cms/djangoapps/contentstore/api/tests/test_validation.py @@ -15,8 +15,8 @@ from rest_framework import status from rest_framework.test import APITestCase from rest_framework.test import APIClient -from openedx.core.djangoapps.authz.tests.mixins import CourseAuthzTestMixin -from openedx_authz.constants.roles import COURSE_STAFF, COURSE_DATA_RESEARCHER +from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin, CourseAuthzTestMixin +from openedx_authz.constants.roles import COURSE_STAFF, COURSE_DATA_RESEARCHER, COURSE_EDITOR from common.djangoapps.course_modes.models import CourseMode from common.djangoapps.course_modes.tests.factories import CourseModeFactory @@ -249,7 +249,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): """ @@ -355,3 +355,86 @@ 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) + + + + +class TestMigrationViewSetCreateAuthz( + CourseAuthoringAuthzTestMixin, + SharedModuleStoreTestCase, + APITestCase, +): + """ + AuthZ tests for: + /api/courses/v1/migrate_legacy_content_blocks// + """ + + @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()) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertIsInstance(response.json(), list) + + def test_unauthorized_user_cannot_list_blocks(self): + """Unauthorized user should receive 403.""" + response = self.unauthorized_client.get(self.url()) + + self.assertEqual(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()) + + self.assertEqual(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()) + + self.assertIn(response.status_code, [status.HTTP_200_OK, status.HTTP_201_CREATED]) diff --git a/cms/djangoapps/contentstore/api/views/course_validation.py b/cms/djangoapps/contentstore/api/views/course_validation.py index fd3e447b2746..2192b15c04cf 100644 --- a/cms/djangoapps/contentstore/api/views/course_validation.py +++ b/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. diff --git a/cms/djangoapps/contentstore/rest_api/v1/views/course_index.py b/cms/djangoapps/contentstore/rest_api/v1/views/course_index.py index 42b5b1e9d78d..d2bbec5cb56d 100644 --- a/cms/djangoapps/contentstore/rest_api/v1/views/course_index.py +++ b/cms/djangoapps/contentstore/rest_api/v1/views/course_index.py @@ -10,6 +10,8 @@ from rest_framework.views import APIView from rest_framework.fields import BooleanField +from openedx_authz.constants.permissions import COURSES_VIEW_COURSE + from cms.djangoapps.contentstore.config.waffle import CUSTOM_RELATIVE_DATES from cms.djangoapps.contentstore.rest_api.v1.mixins import ContainerHandlerMixin from cms.djangoapps.contentstore.rest_api.v1.serializers import ( @@ -25,7 +27,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 @@ -104,7 +106,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({ diff --git a/cms/djangoapps/contentstore/rest_api/v1/views/tests/test_course_index.py b/cms/djangoapps/contentstore/rest_api/v1/views/tests/test_course_index.py index 310d0ba80adc..b9b89698e654 100644 --- a/cms/djangoapps/contentstore/rest_api/v1/views/tests/test_course_index.py +++ b/cms/djangoapps/contentstore/rest_api/v1/views/tests/test_course_index.py @@ -7,6 +7,7 @@ from rest_framework import status from edx_toggles.toggles.testutils import override_waffle_flag +from openedx_authz.constants.roles import COURSE_EDITOR from cms.djangoapps.contentstore.config.waffle import CUSTOM_RELATIVE_DATES from cms.djangoapps.contentstore.rest_api.v1.mixins import PermissionAccessMixin @@ -16,6 +17,7 @@ from common.djangoapps.student.tests.factories import UserFactory from openedx.core.djangoapps.waffle_utils.testutils import WAFFLE_TABLES from xmodule.modulestore.tests.factories import BlockFactory, check_mongo_calls +from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin class CourseIndexViewTest(CourseTestCase, PermissionAccessMixin): @@ -163,3 +165,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) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertIn("course_structure", response.data) + + def test_unauthorized_user_cannot_access_course_index(self): + """Unauthorized user should receive 403.""" + response = self.unauthorized_client.get(self.url) + + self.assertEqual(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) + self.assertEqual(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) + self.assertEqual(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) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertIn("course_structure", response.data) + + def test_superuser_can_access_without_authz_role(self): + """Superuser should access without AuthZ role.""" + response = self.super_client.get(self.url) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertIn("course_structure", response.data) diff --git a/cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py b/cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py index 16dc977c8af8..f7e3c4e4d47e 100644 --- a/cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py +++ b/cms/djangoapps/contentstore/rest_api/v2/views/downstreams.py @@ -96,6 +96,7 @@ from rest_framework.views import APIView from itertools import chain from xblock.core import XBlock +from openedx_authz.constants.permissions import COURSES_VIEW_COURSE from cms.djangoapps.contentstore.models import ComponentLink, ContainerLink, EntityLinkBase from cms.djangoapps.contentstore.rest_api.v2.serializers import ( @@ -120,6 +121,7 @@ view_auth_classes, ) from openedx.core.djangoapps.content_libraries import api as lib_api +from openedx.core.djangoapps.authz.decorators import LegacyAuthoringPermission, user_has_course_permission from xmodule.modulestore.django import modulestore from xmodule.modulestore.exceptions import ItemNotFoundError from openedx.core.djangoapps.video_config.transcripts_utils import clear_transcripts @@ -305,7 +307,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 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 f82859f74bd3..5f826f24f63e 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 @@ -8,11 +8,13 @@ import ddt from django.conf import settings from django.urls import reverse +from rest_framework import status from freezegun import freeze_time from opaque_keys.edx.keys import ContainerKey, UsageKey from opaque_keys.edx.locator import LibraryLocatorV2, LibraryUsageLocatorV2 from openedx_content import models_api as content_models from organizations.models import Organization +from openedx_authz.constants.roles import COURSE_EDITOR from cms.djangoapps.contentstore.helpers import StaticFileNotices from cms.djangoapps.contentstore.tests.utils import CourseTestCase @@ -23,6 +25,7 @@ from common.djangoapps.student.roles import CourseStaffRole from common.djangoapps.student.tests.factories import UserFactory from openedx.core.djangoapps.content_libraries import api as lib_api +from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin from xmodule.modulestore.django import modulestore from xmodule.modulestore.tests.django_utils import ImmediateOnCommitMixin, SharedModuleStoreTestCase from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory @@ -1517,6 +1520,68 @@ def test_200_summary(self): self.assertListEqual(data, expected) +class GetDownstreamSummaryAuthzViewTest( + CourseAuthoringAuthzTestMixin, + _BaseDownstreamViewTestMixin, + ImmediateOnCommitMixin, + SharedModuleStoreTestCase, +): + """ + AuthZ tests for: + GET /api/contentstore/v2/downstreams//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)) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertIsInstance(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)) + + self.assertEqual(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)) + self.assertEqual(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)) + self.assertEqual(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)) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertIsInstance(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)) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertIsInstance(response.json(), list) + + class GetDownstreamDeletedUpstream( _BaseDownstreamViewTestMixin, ImmediateOnCommitMixin, diff --git a/openedx/core/djangoapps/authz/tests/mixins.py b/openedx/core/djangoapps/authz/tests/mixins.py index a8638b5fe375..0bb28dff07e8 100644 --- a/openedx/core/djangoapps/authz/tests/mixins.py +++ b/openedx/core/djangoapps/authz/tests/mixins.py @@ -55,6 +55,14 @@ def setUp(self): self.unauthorized_client = APIClient() self.unauthorized_client.force_authenticate(user=self.unauthorized_user) + self.super_user = UserFactory(is_superuser=True, password=self.password) + self.super_client = APIClient() + self.super_client.force_authenticate(user=self.super_user) + + self.staff_user = UserFactory(is_staff=True, password=self.password) + self.staff_client = APIClient() + self.staff_client.force_authenticate(user=self.staff_user) + def tearDown(self): super().tearDown() AuthzEnforcer.get_enforcer().clear_policy()