feat: authorize object tag updates via openedx-authz when flag is enabled

Add courses.manage_tags permission check on PUT /object_tags/{course_id}/
behind the AUTHZ_COURSE_AUTHORING_FLAG feature flag. When the flag is
enabled for a course, the endpoint enforces the permission via
openedx-authz. When the flag is off, the existing legacy django-rules
checks are used.

Co-authored-by: Kiro <[email protected]>

Author: wgu-taylor-payne

openedx/core/djangoapps/content_tagging/auth.py

@@ -6,23 +6,38 @@
 from opaque_keys import InvalidKeyError
 from opaque_keys.edx.keys import CourseKey
 from openedx_authz import api as authz_api
-from openedx_authz.constants.permissions import COURSES_EXPORT_TAGS
+from openedx_authz.constants.permissions import COURSES_EXPORT_TAGS, COURSES_MANAGE_TAGS
 from openedx_tagging import rules as oel_tagging_rules
 
 from openedx.core import toggles as core_toggles
 
 log = logging.getLogger(__name__)
 
 
+def has_manage_tags_access(user, course_key):
+    """
+    Check if the user has access to manage tags on a course (tag/untag content objects).
+
+    When authz is enabled for the course, checks courses.manage_tags via openedx-authz.
+    Otherwise falls back to the legacy permission check.
+    """
+    if core_toggles.AUTHZ_COURSE_AUTHORING_FLAG.is_enabled(course_key):
+        return authz_api.is_user_allowed(
+            user.username, COURSES_MANAGE_TAGS.identifier, str(course_key)
+        )
+
+    # Legacy: check via django-rules
+    return user.has_perm(
+        "oel_tagging.change_objecttag",
+        oel_tagging_rules.ObjectTagPermissionItem(taxonomy=None, object_id=str(course_key)),
+    )
+
+
 def has_view_object_tags_access(user, object_id):
     """
     Check if the user has access to view object tags for the given object.
     """
-    # If authz is enabled, check for the export tags authz permission
     course_key = None
-    # Try to parse the object_id as a CourseKey, if it fails,
-    # it means object_id is not a course, so we don't validate against authz
-    # and fallback to the legacy check.
     try:
         course_key = CourseKey.from_string(object_id)
     except InvalidKeyError:
@@ -33,9 +48,7 @@ def has_view_object_tags_access(user, object_id):
             user.username, COURSES_EXPORT_TAGS.identifier, str(course_key)
         )
 
-    # Always check for tagging permissions
     return user.has_perm(
         "oel_tagging.view_objecttag",
-        # 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]
     )

openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py

@@ -2136,6 +2136,63 @@ 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 update endpoint with openedx-authz.
+
+    When the AUTHZ_COURSE_AUTHORING_FLAG is enabled for a course,
+    PUT /object_tags/{course_id}/ should enforce courses.manage_tags.
+    """
+
+    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 test_update_object_tags_authorized(self):
+        """Authorized user can update object tags."""
+        url = OBJECT_TAG_UPDATE_URL.format(object_id=self.course_key)
+        resp = self.authorized_client.put(
+            url,
+            {"tagsData": [{"taxonomy": self.taxonomy.id, "tags": ["Tag 1"]}]},
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_update_object_tags_unauthorized(self):
+        """Unauthorized user cannot update object tags."""
+        url = OBJECT_TAG_UPDATE_URL.format(object_id=self.course_key)
+        resp = self.unauthorized_client.put(
+            url,
+            {"tagsData": [{"taxonomy": self.taxonomy.id, "tags": ["Tag 1"]}]},
+            format="json",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_update_object_tags_scoped_to_course(self):
+        """Authorization should only apply to the assigned 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",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+
 @skip_unless_cms
 @ddt.ddt
 class TestDownloadTemplateView(APITestCase):

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

@@ -5,6 +5,7 @@
 
 from django.db.models import Count
 from django.http import StreamingHttpResponse
+from opaque_keys.edx.keys import CourseKey
 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
@@ -15,6 +16,7 @@
 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 (
@@ -26,8 +28,9 @@
     get_unassigned_taxonomies,
     set_taxonomy_orgs,
 )
-from ...auth import has_view_object_tags_access
+from ...auth import has_manage_tags_access, 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 +157,57 @@ 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 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 = self._get_course_key(object_id)
+
+        if course_key and core_toggles.AUTHZ_COURSE_AUTHORING_FLAG.is_enabled(course_key):
+            can_tag = has_manage_tags_access(request.user, course_key)
+            for obj_data in response.data.values():
+                for taxonomy_entry in obj_data.get("taxonomies", []):
+                    taxonomy_entry["can_tag_object"] = can_tag
+
+        return response
+
     def update(self, request, *args, **kwargs) -> Response:
         """
-        Extend the update method to fire CONTENT_OBJECT_ASSOCIATIONS_CHANGED event
+        Extend the update method to check authz permissions and fire events.
         """
+        object_id = kwargs.get('object_id', '')
+        course_key = self._get_course_key(object_id)
+
+        # When authz is enabled for this course, enforce manage_tags permission.
+        # This covers both course-level and block-level tagging within a course.
+        # When authz is not enabled, fall through to the parent's legacy checks.
+        if course_key and core_toggles.AUTHZ_COURSE_AUTHORING_FLAG.is_enabled(course_key):
+            if not has_manage_tags_access(request.user, course_key):
+                raise PermissionDenied(
+                    "You do not have permission to manage tags for this course."
+                )
+
         response = super().update(request, *args, **kwargs)
         if response.status_code == 200:
-            object_id = kwargs.get('object_id')
-
             # .. event_implemented_name: CONTENT_OBJECT_ASSOCIATIONS_CHANGED
             # .. event_type: org.openedx.content_authoring.content.object.associations.changed.v1
             CONTENT_OBJECT_ASSOCIATIONS_CHANGED.send_event(