feat: authorize tags and taxonomies endpoints via openedx-authz when flag is enabled

Add RBAC permission checks (courses.manage_tags, courses.manage_taxonomies)
to content tagging API endpoints behind AUTHZ_COURSE_AUTHORING_FLAG.
Non-course-scoped taxonomy endpoints check the global flag; course-scoped
object_tags endpoints check per-course. Serializer permission fields
(can_change_taxonomy, can_delete_taxonomy, can_tag_object) are authz-aware
so the frontend needs no changes.

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

Author: wgu-taylor-payne

openedx/core/djangoapps/content_tagging/auth.py

@@ -6,23 +6,81 @@
 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, COURSES_MANAGE_TAXONOMIES
 from openedx_tagging import rules as oel_tagging_rules
 
 from openedx.core import toggles as core_toggles
 
 log = logging.getLogger(__name__)
 
 
+def _is_authz_enabled(course_key=None):
+    """
+    Check if authz is enabled. When course_key is provided, checks per-course/org overrides.
+    When course_key is None, checks the global flag state only.
+    """
+    if course_key:
+        return core_toggles.AUTHZ_COURSE_AUTHORING_FLAG.is_enabled(course_key)
+    return core_toggles.AUTHZ_COURSE_AUTHORING_FLAG.is_enabled()
+
+
+def _user_has_authz_permission_in_any_scope(user, permission_identifier):
+    """
+    Check if the user has the given authz permission in any scope.
+    Used for non-course-scoped endpoints (e.g., taxonomy management).
+    """
+    scopes = authz_api.get_scopes_for_user_and_permission(
+        user.username, permission_identifier
+    )
+    return len(scopes) > 0
+
+
+def has_manage_taxonomies_access(user, course_key=None):
+    """
+    Check if the user has access to manage taxonomies (create, edit, delete, import, export, manage orgs).
+
+    When course_key is provided, checks the flag for that specific course.
+    When course_key is None, checks the global flag state and whether the user
+    has the permission in any scope.
+    Falls back to legacy taxonomy admin check when the flag is off.
+    """
+    if _is_authz_enabled(course_key):
+        if course_key:
+            return authz_api.is_user_allowed(
+                user.username, COURSES_MANAGE_TAXONOMIES.identifier, str(course_key)
+            )
+        return _user_has_authz_permission_in_any_scope(user, COURSES_MANAGE_TAXONOMIES.identifier)
+
+    # Legacy: taxonomy admins and org-level staff can manage taxonomies
+    return oel_tagging_rules.is_taxonomy_admin(user)
+
+
+def has_manage_tags_access(user, course_key=None):
+    """
+    Check if the user has access to manage tags (create, edit, delete, import tags on taxonomies,
+    and tag/untag content objects).
+
+    When course_key is provided, checks the flag for that specific course.
+    When course_key is None, checks the global flag state and whether the user
+    has the permission in any scope.
+    Falls back to legacy taxonomy admin check when the flag is off.
+    """
+    if _is_authz_enabled(course_key):
+        if course_key:
+            return authz_api.is_user_allowed(
+                user.username, COURSES_MANAGE_TAGS.identifier, str(course_key)
+            )
+        return _user_has_authz_permission_in_any_scope(user, COURSES_MANAGE_TAGS.identifier)
+
+    # Legacy: taxonomy admins can manage tags
+    return oel_tagging_rules.is_taxonomy_admin(user)
+
+
 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 +91,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/serializers.py

@@ -12,6 +12,7 @@
 from organizations.models import Organization
 from rest_framework import fields, serializers
 
+from ...auth import _is_authz_enabled, has_manage_tags_access, has_manage_taxonomies_access
 from ...models import TaxonomyOrg
 
 
@@ -65,7 +66,9 @@ def validate(self, attrs: dict) -> dict:
 
 class TaxonomyOrgSerializer(TaxonomySerializer):
     """
-    Serializer for Taxonomy objects inclusing the associated orgs
+    Serializer for Taxonomy objects including the associated orgs.
+
+    Overrides permission fields to be authz-aware when the feature flag is enabled.
     """
 
     orgs = serializers.SerializerMethodField()
@@ -89,6 +92,44 @@ def get_all_orgs(self, obj) -> bool:
                 return True
         return False
 
+    def _get_request_user(self):
+        """Return the current request user, or None."""
+        request = self.context.get('request')
+        return request.user if request else None
+
+    def get_can_change(self, instance) -> bool | None:
+        """
+        Override to check authz manage_taxonomies permission when the flag is globally enabled.
+        Falls back to legacy permission check when the flag is off.
+        """
+        if _is_authz_enabled():
+            user = self._get_request_user()
+            if user:
+                return has_manage_taxonomies_access(user)
+        return super().get_can_change(instance)
+
+    def get_can_delete(self, instance) -> bool | None:
+        """
+        Override to check authz manage_taxonomies permission when the flag is globally enabled.
+        Falls back to legacy permission check when the flag is off.
+        """
+        if _is_authz_enabled():
+            user = self._get_request_user()
+            if user:
+                return has_manage_taxonomies_access(user)
+        return super().get_can_delete(instance)
+
+    def get_can_tag_object(self, instance) -> bool | None:
+        """
+        Override to check authz manage_tags permission when the flag is globally enabled.
+        Falls back to legacy permission check when the flag is off.
+        """
+        if _is_authz_enabled():
+            user = self._get_request_user()
+            if user:
+                return has_manage_tags_access(user)
+        return super().get_can_tag_object(instance)
+
     class Meta:
         model = TaxonomySerializer.Meta.model
         fields = TaxonomySerializer.Meta.fields + ["orgs", "all_orgs"]

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

@@ -16,7 +16,7 @@
 from django.urls import reverse
 from edx_django_utils.cache import RequestCache
 from opaque_keys.edx.locator import BlockUsageLocator, CourseLocator, LibraryCollectionLocator, LibraryContainerLocator
-from openedx_authz.constants.roles import COURSE_STAFF
+from openedx_authz.constants.roles import COURSE_ADMIN, COURSE_STAFF
 from openedx_tagging.models import Tag, Taxonomy
 from openedx_tagging.models.system_defined import SystemDefinedTaxonomy
 from openedx_tagging.rest_api.v1.serializers import TaxonomySerializer
@@ -2106,6 +2106,166 @@ def test_superuser_allowed(self):
         resp = client.get(self.get_url(self.course_key))
         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 
+@skip_unless_cms
+class TestTaxonomyEndpointsWithAuthz(CourseAuthzTestMixin, SharedModuleStoreTestCase, APITestCase):
+    """
+    Tests taxonomy management endpoints with openedx-authz.
+
+    When the AUTHZ_COURSE_AUTHORING_FLAG is globally enabled, taxonomy CRUD
+    endpoints should enforce courses.manage_taxonomies / courses.manage_tags
+    permissions via openedx-authz.
+
+    Uses COURSE_ADMIN role because only course_admin has manage_taxonomies permission.
+    """
+
+    authz_roles_to_assign = [COURSE_ADMIN.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.orgA = Organization.objects.create(name="Organization A", short_name="orgA")
+        self.taxonomy = tagging_api.create_taxonomy(name="Test Taxonomy")
+        tagging_api.set_taxonomy_orgs(self.taxonomy, all_orgs=False, orgs=[self.orgA])
+
+    def test_delete_taxonomy_authorized(self):
+        """Authorized user can delete a taxonomy."""
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.taxonomy.id)
+        resp = self.authorized_client.delete(url)
+        self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
+
+    def test_delete_taxonomy_unauthorized(self):
+        """Unauthorized user cannot delete a taxonomy."""
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.taxonomy.id)
+        resp = self.unauthorized_client.delete(url)
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_update_orgs_authorized(self):
+        """Authorized user can update taxonomy orgs."""
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.taxonomy.id)
+        resp = self.authorized_client.put(url, {"all_orgs": True}, format="json")
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_update_orgs_unauthorized(self):
+        """Unauthorized user cannot update taxonomy orgs."""
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.taxonomy.id)
+        resp = self.unauthorized_client.put(url, {"all_orgs": True}, format="json")
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_create_import_authorized(self):
+        """Authorized user can create a taxonomy via import."""
+        url = TAXONOMY_CREATE_IMPORT_URL
+        file = SimpleUploadedFile(
+            "taxonomy.csv",
+            b"id,value\ntag_1,Tag 1\n",
+            content_type="text/csv",
+        )
+        resp = self.authorized_client.post(
+            url,
+            {"taxonomy_name": "Imported", "taxonomy_description": "", "file": file},
+            format="multipart",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
+
+    def test_create_import_unauthorized(self):
+        """Unauthorized user cannot create a taxonomy via import."""
+        url = TAXONOMY_CREATE_IMPORT_URL
+        file = SimpleUploadedFile(
+            "taxonomy.csv",
+            b"id,value\ntag_1,Tag 1\n",
+            content_type="text/csv",
+        )
+        resp = self.unauthorized_client.post(
+            url,
+            {"taxonomy_name": "Imported", "taxonomy_description": "", "file": file},
+            format="multipart",
+        )
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_taxonomy_list_permission_fields_authorized(self):
+        """Authorized user sees correct permission fields in taxonomy list."""
+        url = TAXONOMY_ORG_LIST_URL
+        resp = self.authorized_client.get(url)
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        for taxonomy in resp.data["results"]:
+            self.assertTrue(taxonomy["can_change_taxonomy"])
+            self.assertTrue(taxonomy["can_delete_taxonomy"])
+            self.assertTrue(taxonomy["can_tag_object"])
+
+    def test_taxonomy_list_permission_fields_unauthorized(self):
+        """Unauthorized user sees restricted permission fields in taxonomy list."""
+        url = TAXONOMY_ORG_LIST_URL
+        resp = self.unauthorized_client.get(url)
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        for taxonomy in resp.data["results"]:
+            self.assertFalse(taxonomy["can_change_taxonomy"])
+            self.assertFalse(taxonomy["can_delete_taxonomy"])
+            self.assertFalse(taxonomy["can_tag_object"])
+
+
+@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):