Add listener to m2m_changed signal to invalidate permissions cache if group assignment of a user is changed

fsbraun · fsbraun · commit fc59f73d47b0 · 2024-07-21T12:56:03.000+02:00
diff --git a/cms/signals/__init__.py b/cms/signals/__init__.py
@@ -27,6 +27,7 @@
     pre_save_group,
     pre_save_pagepermission,
     pre_save_user,
+    user_m2m_changed,
 )
 from cms.utils.conf import get_cms_setting
 
@@ -89,6 +90,7 @@ def check_v4_confirmation(**kwargs):
     signals.pre_save.connect(pre_save_user, sender=User, dispatch_uid='cms_pre_save_user')
     signals.post_save.connect(post_save_user, sender=User, dispatch_uid='cms_post_save_user')
     signals.pre_delete.connect(pre_delete_user, sender=User, dispatch_uid='cms_pre_delete_user')
+    signals.m2m_changed.connect(user_m2m_changed, sender=User.groups.through, dispatch_uid='cms_user_m2m_changed')
 
     signals.pre_save.connect(pre_save_user, sender=PageUser, dispatch_uid='cms_pre_save_pageuser')
     signals.pre_delete.connect(pre_delete_user, sender=PageUser, dispatch_uid='cms_pre_delete_pageuser')
diff --git a/cms/signals/permissions.py b/cms/signals/permissions.py
@@ -1,13 +1,19 @@
+from django.contrib.auth import get_user_model
+
 from cms.cache.permissions import clear_user_permission_cache
 from cms.models import PageUser, PageUserGroup
 from menus.menu_pool import menu_pool
 
+User = get_user_model()
+
 
 def post_save_user(instance, raw, created, **kwargs):
     """Signal called when new user is created, required only when CMS_PERMISSION.
     Assigns creator of the user to PageUserInfo model, so we know who had created
     this user account.
 
+    Flushes permission cache for the user.
+
     requires: CurrentUserMiddleware
     """
     from cms.utils.permissions import get_current_user
@@ -21,6 +27,8 @@ def post_save_user(instance, raw, created, **kwargs):
     page_user.__dict__.update(instance.__dict__)
     page_user.save()
 
+    clear_user_permission_cache(instance
+                                )
 
 def post_save_user_group(instance, raw, created, **kwargs):
     """The same like post_save_user, but for Group, required only when
@@ -62,6 +70,11 @@ def pre_delete_group(instance, **kwargs):
         clear_user_permission_cache(user)
 
 
+def user_m2m_changed(instance, action, reverse, model, pk_set, sender, **kwargs):
+    if action in ('pre_add', 'pre_remove',) and not reverse:
+        clear_user_permission_cache(instance)
+
+
 def _clear_users_permissions(instance):
     if instance.user:
         clear_user_permission_cache(instance.user)
diff --git a/cms/tests/test_page_admin.py b/cms/tests/test_page_admin.py
@@ -2647,7 +2647,7 @@ def test_user_cant_delete_page_view_restrictions(self):
 
     def test_permissions_cache_invalidation(self):
         """
-        Test permission cache clearing on page save
+        Permission cache clears on page save
         """
         page = self.get_permissions_test_page()
         staff_user = self.get_staff_user_with_std_permissions()
@@ -2660,6 +2660,42 @@ def test_permissions_cache_invalidation(self):
             self.client.post(endpoint, data)
         self.assertIsNone(get_permission_cache(staff_user, "change_page"))
 
+    def test_permission_cache_invalidation_on_group_add(self):
+        """
+        Permissions cache is invalidated if the group relationship of a user is changed
+        """
+        from django.contrib.auth.models import Group
+
+        page = self.get_permissions_test_page()
+        staff_user = self.get_staff_user_with_std_permissions()
+        set_permission_cache(staff_user, "change_page", [page.pk])
+
+        group = Group(name="test_group")
+        group.save()
+        staff_user.groups.add(group)
+
+        self.assertIsNone(get_permission_cache(staff_user, "change_page"))
+
+
+    def test_permission_cache_invalidation_on_group_remove(self):
+        """
+        Permissions cache is invalidated if the group relationship of a user is changed
+        """
+        from django.contrib.auth.models import Group
+
+        page = self.get_permissions_test_page()
+        staff_user = self.get_staff_user_with_std_permissions()
+        group = Group(name="test_group")
+        group.save()
+        staff_user.groups.add(group)
+
+        set_permission_cache(staff_user, "change_page", [page.pk])
+
+        staff_user.groups.remove(group)
+
+        self.assertIsNone(get_permission_cache(staff_user, "change_page"))
+
+
     def test_user_can_copy_page(self):
         """
         Test that a page can be copied via the admin
