from celery.signals import task_postrun
from celery.signals import task_prerun
from django.conf import settings
+from django.contrib.auth.models import Group
from django.contrib.auth.models import User
from django.db import DatabaseError
from django.db import close_old_connections
from documents.models import PaperlessTask
from documents.models import SavedView
from documents.models import Tag
+from documents.models import UiSettings
from documents.models import Workflow
from documents.models import WorkflowAction
from documents.models import WorkflowRun
)
+@receiver(models.signals.post_delete, sender=User)
+@receiver(models.signals.post_delete, sender=Group)
+def cleanup_user_deletion(sender, instance: User | Group, **kwargs):
+ """
+ When a user or group is deleted, remove non-cascading references.
+ At the moment, just the default permission settings in UiSettings.
+ """
+ # Remove the user permission settings e.g.
+ # DEFAULT_PERMS_OWNER: 'general-settings:permissions:default-owner',
+ # DEFAULT_PERMS_VIEW_USERS: 'general-settings:permissions:default-view-users',
+ # DEFAULT_PERMS_VIEW_GROUPS: 'general-settings:permissions:default-view-groups',
+ # DEFAULT_PERMS_EDIT_USERS: 'general-settings:permissions:default-edit-users',
+ # DEFAULT_PERMS_EDIT_GROUPS: 'general-settings:permissions:default-edit-groups',
+ for ui_settings in UiSettings.objects.all():
+ try:
+ permissions = ui_settings.settings.get("permissions", {})
+ updated = False
+ if isinstance(instance, User):
+ if permissions.get("default_owner") == instance.pk:
+ permissions["default_owner"] = None
+ updated = True
+ if instance.pk in permissions.get("default_view_users", []):
+ permissions["default_view_users"].remove(instance.pk)
+ updated = True
+ if instance.pk in permissions.get("default_change_users", []):
+ permissions["default_change_users"].remove(instance.pk)
+ updated = True
+ elif isinstance(instance, Group):
+ if instance.pk in permissions.get("default_view_groups", []):
+ permissions["default_view_groups"].remove(instance.pk)
+ updated = True
+ if instance.pk in permissions.get("default_change_groups", []):
+ permissions["default_change_groups"].remove(instance.pk)
+ updated = True
+ if updated:
+ ui_settings.settings["permissions"] = permissions
+ ui_settings.save(update_fields=["settings"])
+ except Exception as e:
+ logger.error(
+ f"Error while cleaning up user {instance.pk} ({instance.username}) from ui_settings: {e}"
+ if isinstance(instance, User)
+ else f"Error while cleaning up group {instance.pk} ({instance.name}) from ui_settings: {e}",
+ )
+
+
def add_to_index(sender, document, **kwargs):
from documents import index
from django.test import TestCase
from django.test import override_settings
+from documents.models import UiSettings
from paperless.signals import handle_failed_login
from paperless.signals import handle_social_account_updated
sociallogin=sociallogin,
)
self.assertEqual(list(user.groups.all()), [])
+
+
+class TestUserGroupDeletionCleanup(TestCase):
+ """
+ Test that when a user or group is deleted, references are cleaned up properly
+ from ui_settings
+ """
+
+ def test_user_group_deletion_cleanup(self):
+ """
+ GIVEN:
+ - Existing user
+ - Existing group
+ WHEN:
+ - The user is deleted
+ - The group is deleted
+ THEN:
+ - References in ui_settings are cleaned up
+ """
+ user = User.objects.create_user(username="testuser")
+ user2 = User.objects.create_user(username="testuser2")
+ group = Group.objects.create(name="testgroup")
+
+ ui_settings = UiSettings.objects.create(
+ user=user,
+ settings={
+ "permissions": {
+ "default_owner": user2.id,
+ "default_view_users": [user2.id],
+ "default_change_users": [user2.id],
+ "default_view_groups": [group.id],
+ "default_change_groups": [group.id],
+ },
+ },
+ )
+
+ user2.delete()
+ ui_settings.refresh_from_db()
+ permissions = ui_settings.settings.get("permissions", {})
+ self.assertIsNone(permissions.get("default_owner"))
+ self.assertEqual(permissions.get("default_view_users"), [])
+ self.assertEqual(permissions.get("default_change_users"), [])
+
+ group.delete()
+ ui_settings.refresh_from_db()
+ permissions = ui_settings.settings.get("permissions", {})
+ self.assertEqual(permissions.get("default_view_groups"), [])
+ self.assertEqual(permissions.get("default_change_groups"), [])
+
+ def test_user_group_deletion_error_handling(self):
+ """
+ GIVEN:
+ - Existing user and group
+ WHEN:
+ - The user is deleted and an error occurs during the signal handling
+ THEN:
+ - Error is logged and the system remains stable
+ """
+ user = User.objects.create_user(username="testuser")
+ user2 = User.objects.create_user(username="testuser2")
+ user2_id = user2.id
+ Group.objects.create(name="testgroup")
+
+ UiSettings.objects.create(
+ user=user,
+ ) # invalid, no settings, this probably should not happen in production
+
+ with self.assertLogs("paperless.handlers", level="ERROR") as cm:
+ user2.delete()
+ self.assertIn(
+ f"Error while cleaning up user {user2_id}",
+ cm.output[0],
+ )