diff --git a/cms/djangoapps/contentstore/views/block.py b/cms/djangoapps/contentstore/views/block.py index e606477ee222..4d60640cb2cf 100644 --- a/cms/djangoapps/contentstore/views/block.py +++ b/cms/djangoapps/contentstore/views/block.py @@ -12,6 +12,7 @@ from django.views.decorators.clickjacking import xframe_options_exempt from django.views.decorators.http import require_http_methods from opaque_keys.edx.keys import CourseKey +from openedx_authz.constants.permissions import COURSES_VIEW_COURSE from web_fragments.fragment import Fragment from cms.djangoapps.contentstore.utils import load_services_for_studio @@ -27,6 +28,8 @@ from common.djangoapps.edxmako.shortcuts import render_to_response, render_to_string from common.djangoapps.student.auth import has_studio_read_access, has_studio_write_access from common.djangoapps.util.json_request import JsonResponse, expect_json +from openedx.core.djangoapps.authz.constants import LegacyAuthoringPermission +from openedx.core.djangoapps.authz.decorators import user_has_course_permission from openedx.core.djangoapps.content_tagging.toggles import is_tagging_feature_disabled from openedx.core.lib.xblock_utils import hash_resource, request_token, wrap_xblock, wrap_xblock_aside from xmodule.modulestore.django import modulestore # lint-amnesty, pylint: disable=wrong-import-order @@ -329,7 +332,12 @@ def xblock_outline_handler(request, usage_key_string): a course. """ usage_key = usage_key_with_run(usage_key_string) - if not has_studio_read_access(request.user, usage_key.course_key): + if not user_has_course_permission( + request.user, + COURSES_VIEW_COURSE.identifier, + usage_key.course_key, + LegacyAuthoringPermission.READ, + ): raise PermissionDenied() response_format = request.GET.get("format", "html") diff --git a/cms/djangoapps/contentstore/views/tests/test_block.py b/cms/djangoapps/contentstore/views/tests/test_block.py index bb1206169189..dbaa91861a2d 100644 --- a/cms/djangoapps/contentstore/views/tests/test_block.py +++ b/cms/djangoapps/contentstore/views/tests/test_block.py @@ -19,10 +19,12 @@ from opaque_keys.edx.asides import AsideUsageKeyV2 from opaque_keys.edx.keys import CourseKey, UsageKey from opaque_keys.edx.locator import BlockUsageLocator, CourseLocator +from openedx_authz.constants.roles import COURSE_STAFF from openedx_events.content_authoring.data import DuplicatedXBlockData from openedx_events.content_authoring.signals import XBLOCK_DUPLICATED from openedx_events.testing import OpenEdxEventsTestMixin from pytz import UTC +from rest_framework.test import APITestCase from web_fragments.fragment import Fragment from webob import Response from xblock.core import XBlockAside @@ -54,14 +56,20 @@ from common.djangoapps.xblock_django.user_service import DjangoXBlockUserService from common.test.utils import assert_dict_contains_subset from lms.djangoapps.lms_xblock.mixin import NONSENSICAL_ACCESS_RESTRICTION +from openedx.core.djangoapps.authz.tests.mixins import CourseAuthzTestMixin from openedx.core.djangoapps.content_tagging import api as tagging_api from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration from openedx.core.djangoapps.video_config.toggles import PUBLIC_VIDEO_SHARE +from openedx.core.djangolib.testing.utils import skip_unless_cms from xmodule.course_block import DEFAULT_START_DATE from xmodule.modulestore import ModuleStoreEnum from xmodule.modulestore.django import modulestore from xmodule.modulestore.exceptions import ItemNotFoundError -from xmodule.modulestore.tests.django_utils import TEST_DATA_SPLIT_MODULESTORE, ModuleStoreTestCase +from xmodule.modulestore.tests.django_utils import ( + TEST_DATA_SPLIT_MODULESTORE, + ModuleStoreTestCase, + SharedModuleStoreTestCase, +) from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory, LibraryFactory, check_mongo_calls from xmodule.partitions.partitions import ( ENROLLMENT_TRACK_PARTITION_ID, @@ -4615,3 +4623,60 @@ def test_xblock_edit_view_contains_resources(self): self.assertGreater(len(resource_links), 0, f"No CSS resources found in HTML. Found: {resource_links}") # noqa: PT009 # pylint: disable=line-too-long self.assertGreater(len(script_sources), 0, f"No JS resources found in HTML. Found: {script_sources}") # noqa: PT009 # pylint: disable=line-too-long + + + +@skip_unless_cms +class XBlockOutlineHandlerAuthzTest(CourseAuthzTestMixin, SharedModuleStoreTestCase, APITestCase): + """ + Tests xblock_outline_handler authorization via openedx-authz. + + When the AUTHZ_COURSE_AUTHORING_FLAG is enabled, the endpoint should + enforce courses.view_course via openedx-authz instead of legacy + has_studio_read_access. + """ + + authz_roles_to_assign = [COURSE_STAFF.external_key] + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.password = 'test' + cls.course = CourseFactory.create() + cls.course_key = cls.course.id + cls.staff = StaffFactory(course_key=cls.course_key, password=cls.password) + cls.chapter = BlockFactory.create( + parent_location=cls.course.location, + category="chapter", + display_name="Week 1", + user_id=cls.staff.id, + ) + + def _get_outline_url(self): + return reverse_usage_url("xblock_outline_handler", self.chapter.location) + + def test_authorized_user_can_access(self): + """User with COURSE_STAFF role can access.""" + self.authorized_client.login(username=self.authorized_user.username, password=self.password) + resp = self.authorized_client.get(self._get_outline_url(), HTTP_ACCEPT="application/json") + assert resp.status_code == 200 + + def test_unauthorized_user_cannot_access(self): + """User without role cannot access.""" + self.unauthorized_client.login(username=self.unauthorized_user.username, password=self.password) + resp = self.unauthorized_client.get(self._get_outline_url(), HTTP_ACCEPT="application/json") + assert resp.status_code == 403 + + def test_role_scoped_to_course(self): + """Authorization should only apply to the assigned course.""" + other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.staff.id) + other_chapter = BlockFactory.create( + parent_location=other_course.location, + category="chapter", + display_name="Other Week", + user_id=self.staff.id, + ) + url = reverse_usage_url("xblock_outline_handler", other_chapter.location) + self.authorized_client.login(username=self.authorized_user.username, password=self.password) + resp = self.authorized_client.get(url, HTTP_ACCEPT="application/json") + assert resp.status_code == 403 diff --git a/openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py b/openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py index 35f209f8a7e1..696530ec139c 100644 --- a/openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py +++ b/openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py @@ -1979,19 +1979,19 @@ def test_get_copied_tags(self): assert response.data[str(object_id_2)]["taxonomies"] == expected_tags @ddt.data( - ('staff', 'courseA', 8), + ('staff', 'courseA', 10), ('staff', 'libraryA', 17), ('staff', 'collection_key', 17), - ("content_creatorA", 'courseA', 18, False), + ("content_creatorA", 'courseA', 20, False), ("content_creatorA", 'libraryA', 23, False), ("content_creatorA", 'collection_key', 23, False), ("library_staffA", 'libraryA', 23, False), # Library users can only view objecttags, not change them? ("library_staffA", 'collection_key', 23, False), ("library_userA", 'libraryA', 23, False), ("library_userA", 'collection_key', 23, False), - ("instructorA", 'courseA', 18), - ("course_instructorA", 'courseA', 18), - ("course_staffA", 'courseA', 18), + ("instructorA", 'courseA', 20), + ("course_instructorA", 'courseA', 20), + ("course_staffA", 'courseA', 20), ) @ddt.unpack def test_object_tags_query_count( @@ -2136,6 +2136,112 @@ def test_superuser_allowed(self): resp = client.get(self.get_url(self.course_key)) self.assertEqual(resp.status_code, status.HTTP_200_OK) # noqa: PT009 +@skip_unless_cms +class TestObjectTagUpdateWithAuthz(CourseAuthzTestMixin, SharedModuleStoreTestCase, APITestCase): + """ + Tests object tag endpoints with openedx-authz. + + When the AUTHZ_COURSE_AUTHORING_FLAG is enabled for a course, + PUT /object_tags/{object_id}/ should enforce courses.manage_tags, + and GET /object_tags/{object_id}/ should return authz-aware + can_tag_object values. + + When authz is active, the parent's legacy permission checks + (ObjectTagObjectPermissions and per-taxonomy can_tag_object) are + bypassed entirely — permissions are enforced solely via openedx-authz. + """ + + authz_roles_to_assign = [COURSE_STAFF.external_key] + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.password = 'test' + cls.course = CourseFactory.create() + cls.course_key = cls.course.id + cls.staff = StaffFactory(course_key=cls.course_key, password=cls.password) + + def setUp(self): + super().setUp() + self.taxonomy = tagging_api.create_taxonomy(name="Test Taxonomy") + tagging_api.set_taxonomy_orgs(self.taxonomy, all_orgs=True, orgs=[]) + Tag.objects.create(taxonomy=self.taxonomy, value="Tag 1") + + def _put_tags(self, client): + """Helper to PUT tags on the course.""" + url = OBJECT_TAG_UPDATE_URL.format(object_id=self.course_key) + return client.put( + url, + {"tagsData": [{"taxonomy": self.taxonomy.id, "tags": ["Tag 1"]}]}, + format="json", + ) + + def _make_auditor_client(self, course_key=None): + """ + Create a user with authz course_auditor role (no manage_tags permission). + Since get_permissions() replaces legacy checks with IsAuthenticated when + authz is active, no legacy role is needed to pass dispatch. + """ + from openedx_authz.constants.roles import COURSE_AUDITOR # pylint: disable=import-outside-toplevel + + course_key = course_key or self.course_key + auditor = UserFactory(password=self.password) + self.add_user_to_role_in_course(auditor, COURSE_AUDITOR.external_key, course_key) + client = APIClient() + client.force_authenticate(user=auditor) + return client + + def test_update_object_tags_authorized(self): + """Authorized user can update object tags.""" + assert self._put_tags(self.authorized_client).status_code == status.HTTP_200_OK + + def test_update_object_tags_denied_by_authz(self): + """User with legacy access but no authz manage_tags is denied.""" + auditor_client = self._make_auditor_client() + assert self._put_tags(auditor_client).status_code == status.HTTP_403_FORBIDDEN + + def test_update_object_tags_scoped_to_course(self): + """Authorized user for one course cannot tag objects in another course.""" + other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.staff.id) + url = OBJECT_TAG_UPDATE_URL.format(object_id=other_course.id) + resp = self.authorized_client.put( + url, + {"tagsData": [{"taxonomy": self.taxonomy.id, "tags": ["Tag 1"]}]}, + format="json", + ) + assert resp.status_code == status.HTTP_403_FORBIDDEN + + def test_superuser_allowed(self): + """Superusers should always be allowed.""" + superuser = UserFactory(is_superuser=True) + client = APIClient() + client.force_authenticate(user=superuser) + assert self._put_tags(client).status_code == status.HTTP_200_OK + + def test_retrieve_can_tag_object_authorized(self): + """Authorized user sees can_tag_object=true and can_delete_objecttag=true.""" + self._put_tags(self.authorized_client) # ensure tags exist + url = OBJECT_TAGS_URL.format(object_id=self.course_key) + resp = self.authorized_client.get(url) + assert resp.status_code == status.HTTP_200_OK + for taxonomy_entry in resp.data[str(self.course_key)]["taxonomies"]: + assert taxonomy_entry["can_tag_object"] is True + for tag in taxonomy_entry["tags"]: + assert tag["can_delete_objecttag"] is True + + def test_retrieve_can_tag_object_denied(self): + """User sees can_tag_object=false and can_delete_objecttag=false when authz denies manage_tags.""" + auditor_client = self._make_auditor_client() + self._put_tags(self.authorized_client) # ensure tags exist + url = OBJECT_TAGS_URL.format(object_id=self.course_key) + resp = auditor_client.get(url) + assert resp.status_code == status.HTTP_200_OK + for taxonomy_entry in resp.data[str(self.course_key)]["taxonomies"]: + assert taxonomy_entry["can_tag_object"] is False + for tag in taxonomy_entry["tags"]: + assert tag["can_delete_objecttag"] is False + + @skip_unless_cms @ddt.ddt class TestDownloadTemplateView(APITestCase): diff --git a/openedx/core/djangoapps/content_tagging/rest_api/v1/views.py b/openedx/core/djangoapps/content_tagging/rest_api/v1/views.py index a52a8810b547..60aea6acda1b 100644 --- a/openedx/core/djangoapps/content_tagging/rest_api/v1/views.py +++ b/openedx/core/djangoapps/content_tagging/rest_api/v1/views.py @@ -5,16 +5,23 @@ from django.db.models import Count from django.http import StreamingHttpResponse +from opaque_keys.edx.keys import CourseKey +from openedx_authz import api as authz_api +from openedx_authz.constants.permissions import COURSES_MANAGE_TAGS from openedx_events.content_authoring.data import ContentObjectChangedData, ContentObjectData from openedx_events.content_authoring.signals import CONTENT_OBJECT_ASSOCIATIONS_CHANGED, CONTENT_OBJECT_TAGS_CHANGED from openedx_tagging import rules as oel_tagging_rules +from openedx_tagging.api import TagDoesNotExist, get_object_tags, tag_object +from openedx_tagging.rest_api.v1.serializers import ObjectTagListQueryParamsSerializer, ObjectTagUpdateBodySerializer from openedx_tagging.rest_api.v1.views import ObjectTagView, TaxonomyView from rest_framework import status from rest_framework.decorators import action -from rest_framework.exceptions import PermissionDenied, ValidationError +from rest_framework.exceptions import MethodNotAllowed, PermissionDenied, ValidationError +from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView +from openedx.core import toggles as core_toggles from openedx.core.types.http import RestRequest from ...api import ( @@ -28,6 +35,7 @@ ) from ...auth import has_view_object_tags_access from ...rules import get_admin_orgs +from ...utils import get_context_key_from_key_string from .filters import ObjectTagTaxonomyOrgFilterBackend, UserOrgFilterBackend from .serializers import ( ObjectTagCopiedMinimalSerializer, @@ -154,14 +162,110 @@ class ObjectTagOrgView(ObjectTagView): minimal_serializer_class = ObjectTagCopiedMinimalSerializer filter_backends = [ObjectTagTaxonomyOrgFilterBackend] + def _get_course_key(self, object_id): + """ + Extract the course key from any content key string. + Returns None if the object_id is not course-related (e.g., library content). + """ + try: + context_key = get_context_key_from_key_string(object_id) + if isinstance(context_key, CourseKey): + return context_key + except ValueError: + pass + return None + + def _is_authz_active(self, object_id): + """ + Returns (course_key, True) if authz is enabled for this object's course, + (None, False) otherwise. + """ + course_key = self._get_course_key(object_id) + if course_key and core_toggles.enable_authz_course_authoring(course_key): + return course_key, True + return course_key, False + + def get_permissions(self): + """ + When authz is enabled for the course, skip the parent's legacy + DjangoObjectPermissions. Our update/retrieve methods handle + permissions via openedx-authz directly. + """ + object_id = self.kwargs.get('object_id', '') + _, authz_active = self._is_authz_active(object_id) + if authz_active: + return [IsAuthenticated()] + return super().get_permissions() + + def get_queryset(self): + """ + When authz is active, skip the parent's legacy view_objecttag permission + check inside get_queryset. Any authenticated user can view tags; the + can_tag_object field controls edit access. + """ + object_id = self.kwargs["object_id"] + _, authz_active = self._is_authz_active(object_id) + if authz_active: + query_params = ObjectTagListQueryParamsSerializer( + data=self.request.query_params.dict() + ) + query_params.is_valid(raise_exception=True) + taxonomy = query_params.validated_data.get("taxonomy", None) + taxonomy_id = taxonomy.cast().id if taxonomy else None + + if object_id.endswith("*") or "," in object_id: + raise ValidationError("Retrieving tags from multiple objects is not yet supported.") + + return get_object_tags(object_id, taxonomy_id) + return super().get_queryset() + + def retrieve(self, request, *args, **kwargs): + """ + Override retrieve to update can_tag_object permission fields when authz is enabled. + + The parent serializer computes can_tag_object via legacy django-rules. When authz + is enabled for the course, we override those values so the frontend correctly + shows/hides tagging controls. + """ + response = super().retrieve(request, *args, **kwargs) + object_id = kwargs.get('object_id', '') + course_key, authz_active = self._is_authz_active(object_id) + + if authz_active: + can_tag = authz_api.is_user_allowed( + request.user.username, COURSES_MANAGE_TAGS.identifier, str(course_key) + ) + for obj_data in response.data.values(): + for taxonomy_entry in obj_data.get("taxonomies", []): + taxonomy_entry["can_tag_object"] = can_tag + for tag in taxonomy_entry.get("tags", []): + tag["can_delete_objecttag"] = can_tag + + return response + def update(self, request, *args, **kwargs) -> Response: """ - Extend the update method to fire CONTENT_OBJECT_ASSOCIATIONS_CHANGED event + Update object tags. When authz is enabled for the course, permissions are + enforced entirely via openedx-authz (courses.manage_tags), bypassing the + parent's legacy per-taxonomy permission checks. When authz is not enabled, + delegates to the parent's update which uses legacy django-rules checks. """ - response = super().update(request, *args, **kwargs) - if response.status_code == 200: - object_id = kwargs.get('object_id') + object_id = kwargs.get('object_id', '') + course_key, authz_active = self._is_authz_active(object_id) + + if authz_active: + if not authz_api.is_user_allowed( + request.user.username, COURSES_MANAGE_TAGS.identifier, str(course_key) + ): + raise PermissionDenied( + "You do not have permission to manage tags for this course." + ) + kwargs.pop('object_id', None) + response = self._update_tags(request, object_id, **kwargs) + else: + response = super().update(request, *args, **kwargs) + if response.status_code == 200: # .. event_implemented_name: CONTENT_OBJECT_ASSOCIATIONS_CHANGED # .. event_type: org.openedx.content_authoring.content.object.associations.changed.v1 CONTENT_OBJECT_ASSOCIATIONS_CHANGED.send_event( @@ -180,6 +284,35 @@ def update(self, request, *args, **kwargs) -> Response: return response + def _update_tags(self, request, object_id, **kwargs): + """ + Apply tag updates without legacy permission checks. + Replicates the parent's ObjectTagView.update() logic for tag saving + but skips the per-taxonomy oel_tagging.can_tag_object check. + """ + partial = kwargs.pop('partial', False) + if partial: + raise MethodNotAllowed("PATCH", detail="PATCH not allowed") + + body = ObjectTagUpdateBodySerializer(data=request.data) + body.is_valid(raise_exception=True) + + data = body.validated_data.get("tagsData", []) + if not data: + return self.retrieve(request, object_id) + + for tags_data in data: + taxonomy = tags_data.get("taxonomy") + tags = tags_data.get("tags", []) + try: + tag_object(object_id, taxonomy, tags) + except TagDoesNotExist as e: + raise ValidationError from e + except ValueError as e: + raise ValidationError from e + + return self.retrieve(request, object_id) + class ObjectTagExportView(APIView): """"