feat: check authz permissions for course tagging

Author: wgu-taylor-payne

cms/djangoapps/contentstore/views/block.py

@@ -11,8 +11,8 @@
 from django.utils.translation import gettext as _
 from django.views.decorators.clickjacking import xframe_options_exempt
 from django.views.decorators.http import require_http_methods
-from openedx_authz.constants.permissions import COURSES_VIEW_COURSE
 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
@@ -28,9 +28,9 @@
 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.content_tagging.toggles import is_tagging_feature_disabled
-from openedx.core.djangoapps.authz.decorators import user_has_course_permission
 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
 from xmodule.x_module import (  # lint-amnesty, pylint: disable=wrong-import-order
@@ -332,7 +332,12 @@ def xblock_outline_handler(request, usage_key_string):
     a course.
     """
     usage_key = usage_key_with_run(usage_key_string)
-    if not user_has_course_permission(request.user, COURSES_VIEW_COURSE.identifier, usage_key.course_key, LegacyAuthoringPermission.READ):
+    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

@@ -9,7 +9,6 @@
 import ddt
 from bs4 import BeautifulSoup
 from django.conf import settings
-from django.core.exceptions import PermissionDenied
 from django.http import Http404
 from django.test import TestCase
 from django.test.client import RequestFactory
@@ -20,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
@@ -55,13 +55,10 @@
 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.constants import LegacyAuthoringPermission
 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
-from openedx_authz.constants.permissions import COURSES_VIEW_COURSE
-from openedx_authz.constants.roles import COURSE_STAFF
 from xmodule.course_block import DEFAULT_START_DATE
 from xmodule.modulestore import ModuleStoreEnum
 from xmodule.modulestore.django import modulestore
@@ -3529,10 +3526,10 @@ 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
@@ -3544,21 +3541,21 @@ 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
@@ -3570,10 +3567,10 @@ 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
@@ -3585,10 +3582,10 @@ 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"
@@ -3603,10 +3600,10 @@ 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
 
 

openedx/core/djangoapps/content_tagging/auth.py

@@ -5,12 +5,15 @@
 
 from opaque_keys import InvalidKeyError
 from opaque_keys.edx.keys import CourseKey
+from opaque_keys.edx.locator import LibraryLocatorV2
 from openedx_authz import api as authz_api
 from openedx_authz.constants.permissions import COURSES_EXPORT_TAGS
 from openedx_tagging import rules as oel_tagging_rules
 
 from openedx.core import toggles as core_toggles
 
+from .utils import get_context_key_from_key_string
+
 log = logging.getLogger(__name__)
 
 
@@ -39,3 +42,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 course_key
+    try:
+        context_key = get_context_key_from_key_string(object_id)
+        if not isinstance(context_key, CourseKey) or isinstance(context_key, LibraryLocatorV2):
+            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, context_key
+
+    # 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)