fix: restrict mute privileges to discussion moderators only (#224)

Fixed discussion moderation permissions to restrict delete, ban, mute, and restore operations to discussion moderators only. Course staff and course instructors were incorrectly granted full moderation privileges when mute feature was added - they are authoring roles and should not have discussion moderation access.

Author: Alam-2U

lms/djangoapps/discussion/rest_api/forum_mute_views.py

@@ -21,7 +21,6 @@
 from lms.djangoapps.discussion.rest_api.permissions import (
     CanMuteUsers
 )
-from common.djangoapps.student.roles import CourseStaffRole, CourseInstructorRole, GlobalStaff
 from lms.djangoapps.discussion.rest_api.serializers import (
     MuteRequestSerializer,
     UnmuteRequestSerializer,
@@ -79,12 +78,10 @@ def _get_target_user_and_data(request_data, course_id):
 
 
 def _is_privileged_user(user, course_key):
-    """Check if user has privileged permissions."""
+    """Check if user has discussion moderation privileges."""
     return (
         has_discussion_privileges(user, course_key) or
-        GlobalStaff().has_user(user) or
-        CourseStaffRole(course_key).has_user(user) or
-        CourseInstructorRole(course_key).has_user(user)
+        user.is_staff
     )
 
 

lms/djangoapps/discussion/rest_api/permissions.py

@@ -105,13 +105,15 @@ def get_editable_fields(cc_content: Union[Thread, Comment], context: Dict) -> Se
         # Flagging/un-flagging is always available.
         return {"abuse_flagged"}
 
+    is_author = _is_author(cc_content, context)
+
     # Map each field to the condition in which it's editable.
     editable_fields = {
         "abuse_flagged": True,
         "closed": is_thread and has_moderation_privilege,
         "close_reason_code": is_thread and has_moderation_privilege,
         "pinned": is_thread and (has_moderation_privilege or is_staff_or_admin),
-        "muted": is_thread and (has_moderation_privilege or is_staff_or_admin),
+        "muted": is_thread and has_moderation_privilege and not is_author,
         "read": is_thread,
     }
     if is_thread:
@@ -121,7 +123,6 @@ def get_editable_fields(cc_content: Union[Thread, Comment], context: Dict) -> Se
         # Return only editable fields
         return _filter_fields(editable_fields)
 
-    is_author = _is_author(cc_content, context)
     editable_fields.update({
         "voted": has_moderation_privilege or not is_author or is_staff_or_admin,
         "raw_body": has_moderation_privilege or is_author,
@@ -193,13 +194,6 @@ def can_take_action_on_spam(user, course_id):
     """
     Returns if the user has access to take action against forum spam posts.
 
-    Grants access to:
-    - Global Staff (user.is_staff or GlobalStaff role)
-    - Course Staff for the specific course
-    - Course Instructors for the specific course
-    - Forum Moderators for the specific course
-    - Forum Administrators for the specific course
-
     Parameters:
         user: User object
         course_id: CourseKey or string of course_id
@@ -214,14 +208,6 @@ def can_take_action_on_spam(user, course_id):
     if isinstance(course_id, str):
         course_id = CourseKey.from_string(course_id)
 
-    # Check if user is Course Staff or Instructor for this specific course
-    if CourseStaffRole(course_id).has_user(user):
-        return True
-
-    if CourseInstructorRole(course_id).has_user(user):
-        return True
-
-    # Check forum moderator/administrator roles for this specific course
     user_roles = set(
         Role.objects.filter(
             users=user,
@@ -236,18 +222,6 @@ def can_take_action_on_spam(user, course_id):
 class IsAllowedToBulkDelete(permissions.BasePermission):
     """
     Permission that checks if the user is allowed to perform bulk delete and ban operations.
-
-    Grants access to:
-    - Global Staff (superusers)
-    - Course Staff
-    - Course Instructors
-    - Forum Moderators
-    - Forum Administrators
-
-    Denies access to:
-    - Unauthenticated users
-    - Regular students
-    - Community TAs (they can moderate individual posts but not bulk delete)
     """
 
     def has_permission(self, request, view):
@@ -279,45 +253,39 @@ def has_permission(self, request, view):
 class CanMuteUsers(permissions.BasePermission):
     """
     Permission class for all mute/unmute operations.
-    Handles muting, unmuting, and basic course access permissions.
     """
 
     @staticmethod
     def _is_privileged_user(user, course_id):
         """
-        Check if user has discussion privileges.
+        Check if user has discussion moderation privileges.
         """
         return (
             has_discussion_privileges(user, course_id) or
-            GlobalStaff().has_user(user) or
-            CourseStaffRole(course_id).has_user(user) or
-            CourseInstructorRole(course_id).has_user(user)
+            GlobalStaff().has_user(user)
         )
 
     def has_permission(self, request, view):
-        """Check basic mute permissions - same logic as IsStaffOrCourseTeamOrEnrolled"""
+        """
+        Check if user has permission to access mute/unmute endpoints.
+        """
         if not request.user.is_authenticated:
             return False
 
-        # Get course_id from URL kwargs first (where it's actually passed)
         course_id = view.kwargs.get("course_id")
         if not course_id:
             return False
 
-        # Convert course_id to CourseKey if it's a string
         if isinstance(course_id, str):
             try:
                 course_id = CourseKey.from_string(course_id)
             except InvalidKeyError:
                 return False
 
-        # Use same permission logic as IsStaffOrCourseTeamOrEnrolled
         return (
             GlobalStaff().has_user(request.user) or
-            CourseStaffRole(course_id).has_user(request.user) or
-            CourseInstructorRole(course_id).has_user(request.user) or
-            CourseEnrollment.is_enrolled(request.user, course_id) or
-            has_discussion_privileges(request.user, course_id)
+            has_discussion_privileges(request.user, course_id) or
+            CourseEnrollment.is_enrolled(request.user, course_id)
         )
 
     @staticmethod
@@ -415,19 +383,6 @@ def can_unmute(requesting_user, target_user, course_id, scope='personal'):
 class IsAllowedToRestore(permissions.BasePermission):
     """
     Permission that checks if the user has privileges to restore individual deleted content.
-
-    This permission is intentionally more permissive than IsAllowedToBulkDelete because:
-    - Restoring individual content is a less risky operation than bulk deletion
-    - Users who can see deleted content should be able to restore it
-    - Course-level moderation staff need this capability for day-to-day moderation
-
-    Allowed users (course-level permissions):
-    - Global staff (platform-wide)
-    - Course instructors
-    - Course staff
-    - Discussion moderators (course-specific)
-    - Discussion community TAs (course-specific)
-    - Discussion administrators (course-specific)
     """
 
     def has_permission(self, request, view):
@@ -449,12 +404,6 @@ def has_permission(self, request, view):
         except InvalidKeyError:
             return False
 
-        # Check if user is course staff or instructor
-        if CourseStaffRole(course_key).has_user(request.user) or \
-           CourseInstructorRole(course_key).has_user(request.user):
-            return True
-
-        # Check if user has discussion privileges (moderator, community TA, administrator)
         if has_discussion_privileges(request.user, course_key):
             return True
 

lms/djangoapps/discussion/rest_api/serializers.py

@@ -89,7 +89,7 @@ def get_context(course, request, thread=None):
     cc_requester["course_id"] = course.id
     course_discussion_settings = CourseDiscussionSettings.get(course.id)
     is_global_staff = GlobalStaff().has_user(requester)
-    all_privileged_ids = set(moderator_user_ids) | set(ta_user_ids) | set(course_staff_user_ids)
+    all_privileged_ids = set(moderator_user_ids) | set(ta_user_ids)
     has_moderation_privilege = requester.id in all_privileged_ids or is_global_staff
     return {
         "course": course,
@@ -242,14 +242,22 @@ def _validate_non_updatable(self, value):
 
     def _is_user_privileged(self, user_id):
         """
-        Returns a boolean indicating whether the given user_id identifies a
-        privileged user.
+        Returns a boolean indicating whether the given user_id identifies a privileged user.
         """
-        return (
+        is_privileged = (
             user_id in self.context["moderator_user_ids"]
             or user_id in self.context["ta_user_ids"]
         )
 
+        if not is_privileged:
+            try:
+                user = User.objects.get(id=user_id)
+                is_privileged = GlobalStaff().has_user(user)
+            except User.DoesNotExist:
+                pass
+
+        return is_privileged
+
     def _is_anonymous(self, obj):
         """
         Returns a boolean indicating whether the content should be anonymous to
@@ -271,16 +279,22 @@ def get_author(self, obj):
 
     def _get_user_label(self, user_id):
         """
-        Returns the role label (i.e. "Staff", "Moderator" or "Community TA") for the user
-        with the given id.
+        Returns the role label for the user with the given id.
         """
-        is_staff = user_id in self.context["course_staff_user_ids"]
         is_moderator = user_id in self.context["moderator_user_ids"]
         is_ta = user_id in self.context["ta_user_ids"]
 
+        is_global_staff = False
+        if not (is_moderator or is_ta):
+            try:
+                user = User.objects.get(id=user_id)
+                is_global_staff = GlobalStaff().has_user(user)
+            except User.DoesNotExist:
+                pass
+
         return (
             "Staff"
-            if is_staff
+            if is_global_staff
             else "Moderator" if is_moderator else "Community TA" if is_ta else None
         )
 

lms/djangoapps/discussion/rest_api/tests/test_api_v2.py

@@ -378,7 +378,6 @@ def test_basic_in_blackout_period_with_user_access(self, mock_emit):
                     "closed",
                     "copy_link",
                     "following",
-                    "muted",
                     "pinned",
                     "raw_body",
                     "read",

lms/djangoapps/discussion/rest_api/tests/test_moderation_permissions.py

@@ -35,17 +35,17 @@ def test_global_staff_role_has_permission(self):
         GlobalStaff().add_users(user)
         self.assertTrue(can_take_action_on_spam(user, self.course_key))
 
-    def test_course_staff_has_permission(self):
-        """Course staff should have permission for their course."""
+    def test_course_staff_no_permission(self):
+        """Course staff should NOT have permission (authoring role only)."""
         user = UserFactory.create()
         CourseStaffRole(self.course_key).add_users(user)
-        self.assertTrue(can_take_action_on_spam(user, self.course_key))
+        self.assertFalse(can_take_action_on_spam(user, self.course_key))
 
-    def test_course_instructor_has_permission(self):
-        """Course instructors should have permission for their course."""
+    def test_course_instructor_no_permission(self):
+        """Course instructors should NOT have permission (authoring role only)."""
         user = UserFactory.create()
         CourseInstructorRole(self.course_key).add_users(user)
-        self.assertTrue(can_take_action_on_spam(user, self.course_key))
+        self.assertFalse(can_take_action_on_spam(user, self.course_key))
 
     def test_forum_moderator_has_permission(self):
         """Forum moderators should have permission for their course."""
@@ -74,16 +74,18 @@ def test_community_ta_no_permission(self):
         self.assertFalse(can_take_action_on_spam(user, self.course_key))
 
     def test_staff_different_course_no_permission(self):
-        """Staff from a different course should not have permission."""
+        """Discussion moderators from a different course should not have permission."""
         other_course = CourseFactory.create(org='OtherX', number='CS201', run='2024')
         user = UserFactory.create()
-        CourseStaffRole(other_course.id).add_users(user)
+        role = Role.objects.create(name='Moderator', course_id=other_course.id)
+        role.users.add(user)
         self.assertFalse(can_take_action_on_spam(user, self.course_key))
 
     def test_accepts_string_course_id(self):
         """Function should accept string course_id and convert it."""
         user = UserFactory.create()
-        CourseStaffRole(self.course_key).add_users(user)
+        role = Role.objects.create(name='Moderator', course_id=self.course_key)
+        role.users.add(user)
         self.assertTrue(can_take_action_on_spam(user, str(self.course_key)))
 
 
@@ -130,23 +132,23 @@ def test_global_staff_with_course_id_in_data(self):
 
         self.assertTrue(self.permission.has_permission(request, view))
 
-    def test_course_staff_with_course_id_in_data(self):
-        """Course staff should have permission when course_id is in request data."""
+    def test_course_staff_denied(self):
+        """Course staff should NOT have permission (authoring role only)."""
         user = UserFactory.create()
         CourseStaffRole(self.course.id).add_users(user)
         request = self._create_request_with_data(user, self.course_key)
         view = self._create_view_with_kwargs()
 
-        self.assertTrue(self.permission.has_permission(request, view))
+        self.assertFalse(self.permission.has_permission(request, view))
 
-    def test_course_instructor_with_course_id_in_data(self):
-        """Course instructors should have permission when course_id is in request data."""
+    def test_course_instructor_denied(self):
+        """Course instructors should NOT have permission (authoring role only)."""
         user = UserFactory.create()
         CourseInstructorRole(self.course.id).add_users(user)
         request = self._create_request_with_data(user, self.course_key)
         view = self._create_view_with_kwargs()
 
-        self.assertTrue(self.permission.has_permission(request, view))
+        self.assertFalse(self.permission.has_permission(request, view))
 
     def test_forum_moderator_with_course_id_in_data(self):
         """Forum moderators should have permission when course_id is in request data."""
@@ -169,7 +171,8 @@ def test_regular_student_denied(self):
     def test_course_id_in_url_kwargs(self):
         """Permission should work when course_id is in URL kwargs."""
         user = UserFactory.create()
-        CourseStaffRole(self.course.id).add_users(user)
+        role = Role.objects.create(name='Moderator', course_id=self.course.id)
+        role.users.add(user)
         request = self.factory.get('/api/discussion/v1/moderation/banned-users/')
         request.user = user
         request.data = {}
@@ -193,10 +196,11 @@ def test_no_course_id_only_global_staff_allowed(self):
         self.assertFalse(self.permission.has_permission(request, view))
 
     def test_staff_different_course_denied(self):
-        """Staff from different course should be denied."""
+        """Discussion moderators from different course should be denied."""
         other_course = CourseFactory.create(org='OtherX', number='CS201', run='2024')
         user = UserFactory.create()
-        CourseStaffRole(other_course.id).add_users(user)
+        role = Role.objects.create(name='Moderator', course_id=other_course.id)
+        role.users.add(user)
         request = self._create_request_with_data(user, self.course_key)
         view = self._create_view_with_kwargs()