feat: enforce openedx-authz permissions on object tag and xblock outline endpoints

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")

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.tests.utils 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
@@ -53,14 +55,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,
@@ -4553,3 +4561,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

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,108 @@ 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 in GET response."""
+        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
+
+    def test_retrieve_can_tag_object_denied(self):
+        """User sees can_tag_object=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
+
+
 @skip_unless_cms
 @ddt.ddt
 class TestDownloadTemplateView(APITestCase):