feat: use authz permission checks for tagging in course authoring

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,6 +19,7 @@
 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
@@ -54,6 +55,7 @@
 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 CourseAuthoringAuthzTestMixin
 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
@@ -3486,6 +3488,125 @@ def validate_xblock_info_consistency(
             self.assertIsNone(xblock_info.get("child_info", None))  # noqa: PT009
 
 
+class TestXBlockOutlineHandlerAuthz(CourseAuthoringAuthzTestMixin, ItemTest):
+    """
+    Unit tests for xblock_outline_handler authorization functionality.
+    """
+
+    def setUp(self):
+        super().setUp()
+        user_id = self.user.id
+        self.chapter = BlockFactory.create(
+            parent_location=self.course.location,
+            category="chapter",
+            display_name="Week 1",
+            user_id=user_id,
+        )
+        self.sequential = BlockFactory.create(
+            parent_location=self.chapter.location,
+            category="sequential",
+            display_name="Lesson 1",
+            user_id=user_id,
+        )
+        self.vertical = BlockFactory.create(
+            parent_location=self.sequential.location,
+            category="vertical",
+            display_name="Unit 1",
+            user_id=user_id,
+        )
+        # Assign COURSE_STAFF role to authorized_user for the course
+        self.add_user_to_role_in_course(
+            self.authorized_user,
+            COURSE_STAFF.external_key,
+            self.course.id
+        )
+
+    def test_authorized_user_gets_json_response(self):
+        """
+        Test that authorized user gets JSON response from xblock_outline_handler.
+        """
+        outline_url = reverse_usage_url("xblock_outline_handler", self.usage_key)
+
+        self.client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.client.get(outline_url, HTTP_ACCEPT="application/json")
+
+        assert resp.status_code == 200
+        json_response = json.loads(resp.content.decode("utf-8"))
+        assert "id" in json_response
+        assert "display_name" in json_response
+        assert "child_info" in json_response
+
+    def test_unauthorized_user_gets_permission_denied(self):
+        """
+        Test that unauthorized user gets 403 response from xblock_outline_handler.
+        """
+        outline_url = reverse_usage_url("xblock_outline_handler", self.usage_key)
+
+        self.client.login(username=self.unauthorized_user.username, password=self.password)
+        resp = self.client.get(outline_url, HTTP_ACCEPT="application/json")
+
+        assert resp.status_code == 403
+
+    def test_superuser_gets_json_response(self):
+        """
+        Test that superuser gets JSON response from xblock_outline_handler.
+        """
+        outline_url = reverse_usage_url("xblock_outline_handler", self.usage_key)
+
+        self.client.login(username=self.super_user.username, password=self.password)
+        resp = self.client.get(outline_url, HTTP_ACCEPT="application/json")
+
+        assert resp.status_code == 200
+        json_response = json.loads(resp.content.decode("utf-8"))
+        assert "id" in json_response
+        assert "display_name" in json_response
+        assert "child_info" in json_response
+
+    def test_staff_user_gets_json_response(self):
+        """
+        Test that staff user gets JSON response from xblock_outline_handler.
+        """
+        outline_url = reverse_usage_url("xblock_outline_handler", self.usage_key)
+
+        self.client.login(username=self.staff_user.username, password=self.password)
+        resp = self.client.get(outline_url, HTTP_ACCEPT="application/json")
+
+        assert resp.status_code == 200
+        json_response = json.loads(resp.content.decode("utf-8"))
+        assert "id" in json_response
+        assert "display_name" in json_response
+        assert "child_info" in json_response
+
+    def test_authorized_chapter_outline(self):
+        """
+        Test that authorized user can access chapter-level outline.
+        """
+        outline_url = reverse_usage_url("xblock_outline_handler", self.chapter.location)
+
+        self.client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.client.get(outline_url, HTTP_ACCEPT="application/json")
+
+        assert resp.status_code == 200
+        json_response = json.loads(resp.content.decode("utf-8"))
+        assert json_response["display_name"] == "Week 1"
+        assert "child_info" in json_response
+        # Verify that children are included (should have the sequential)
+        children = json_response["child_info"]["children"]
+        assert len(children) > 0
+        assert children[0]["display_name"] == "Lesson 1"
+
+    def test_unauthorized_chapter_outline(self):
+        """
+        Test that unauthorized user cannot access chapter-level outline.
+        """
+        outline_url = reverse_usage_url("xblock_outline_handler", self.chapter.location)
+
+        self.client.login(username=self.unauthorized_user.username, password=self.password)
+        resp = self.client.get(outline_url, HTTP_ACCEPT="application/json")
+
+        assert resp.status_code == 403
+
+
 class TestGetMetadataWithProblemDefaults(ModuleStoreTestCase):
     """
     Unit tests for _get_metadata_with_problem_defaults.

openedx/core/djangoapps/content_tagging/auth.py

@@ -11,6 +11,8 @@
 
 from openedx.core import toggles as core_toggles
 
+from .utils import get_context_key_from_key_string
+
 log = logging.getLogger(__name__)
 
 
@@ -39,3 +41,27 @@ def has_view_object_tags_access(user, object_id):
         # The obj arg expects a model, but we are passing an object
         oel_tagging_rules.ObjectTagPermissionItem(taxonomy=None, object_id=object_id),  # type: ignore[arg-type]
     )
+
+
+def should_use_authz_for_object(object_id) -> tuple[bool, CourseKey | None]:
+    """
+    Check if openedx-authz should be used for the given object based on the context key and toggle.
+
+    Returns (should_use_authz, course_key) where:
+    - should_use_authz: True if authz should be used, False otherwise
+    - course_key: The CourseKey if object is a course, None otherwise
+    """
+    # Extract context_key and ensure it is a CourseKey (not a LibraryLocatorV2, etc.)
+    try:
+        context_key = get_context_key_from_key_string(object_id)
+        if not isinstance(context_key, CourseKey):
+            return False, None
+    except (ValueError, AttributeError):
+        return False, None
+
+    # Check if toggle is active
+    if not core_toggles.enable_authz_course_authoring(context_key):
+        return False, None
+
+    # Authz should be used for this course object
+    return True, context_key

openedx/core/djangoapps/content_tagging/rest_api/v1/serializers.py

@@ -4,14 +4,18 @@
 
 from __future__ import annotations
 
+from openedx_authz import api as authz_api
+from openedx_authz.constants.permissions import COURSES_MANAGE_TAGS
 from openedx_tagging.rest_api.v1.serializers import (
     ObjectTagMinimalSerializer,
+    ObjectTagsByTaxonomySerializer,
     TaxonomyListQueryParamsSerializer,
     TaxonomySerializer,
 )
 from organizations.models import Organization
 from rest_framework import fields, serializers
 
+from ...auth import should_use_authz_for_object
 from ...models import TaxonomyOrg
 
 
@@ -95,27 +99,61 @@ class Meta:
         read_only_fields = ["orgs", "all_orgs"]
 
 
+class ObjectTagOrgByTaxonomySerializer(ObjectTagsByTaxonomySerializer):
+    """
+    Extend ObjectTagsByTaxonomySerializer to conditionally use openedx-authz for can_tag_object.
+    """
+
+    def can_tag_object(self, obj_tag) -> bool | None:
+        """
+        Check if the user is authorized to tag the provided object.
+        Conditionally use openedx-authz for course objects with the toggle enabled.
+        """
+        should_use_authz, course_key = should_use_authz_for_object(obj_tag.object_id)
+        if should_use_authz:
+            request = self.context.get('request')
+            if request and hasattr(request, 'user'):
+                return authz_api.is_user_allowed(
+                    request.user.username, COURSES_MANAGE_TAGS.identifier, str(course_key)
+                )
+            return False
+
+        # Fall back to parent implementation
+        return super().can_tag_object(obj_tag)
+
+
 class ObjectTagCopiedMinimalSerializer(ObjectTagMinimalSerializer):
     """
     Serializer for Object Tags.
 
-    This override `get_can_delete_objecttag` to avoid delete
-    object tags if is copied.
+    This overrides `can_delete_object_tag` to avoid deleting
+    object tags if they are copied and to conditionally use openedx-authz.
     """
 
     is_copied = serializers.BooleanField(read_only=True)
 
     class Meta(ObjectTagMinimalSerializer.Meta):
         fields = ObjectTagMinimalSerializer.Meta.fields + ["is_copied"]
 
-    def get_can_delete_objecttag(self, instance):
+    def can_delete_object_tag(self, instance) -> bool | None:
         """
-        Verify if the user can delete the object tag.
+        Check if the user is authorized to delete the provided tag.
 
-        Override to return `False` if the object tag is copied.
+        Override to return `False` if the object tag is copied,
+        and conditionally use openedx-authz for course objects with the toggle enabled.
         """
         if instance.is_copied:
             # The user can't delete copied tags.
             return False
 
-        return super().get_can_delete_objecttag(instance)
+        should_use_authz, course_key = should_use_authz_for_object(instance.object_id)
+        if should_use_authz:
+            request = self.context.get('request')
+            if request and hasattr(request, 'user'):
+                return authz_api.is_user_allowed(
+                    request.user.username, COURSES_MANAGE_TAGS.identifier, str(course_key)
+                )
+            return False
+
+        # Fall back to parent implementation
+        return super().can_delete_object_tag(instance)