Fixed #18763 -- Added with_perm() to User manager.

Co-authored-by: Berker Peksag <berker.peksag@gmail.com>
Co-authored-by: Nick Pope <nick.pope@flightdataservices.com>

django/contrib/auth/backends.py

@@ -3,6 +3,7 @@
 
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Permission
+from django.db.models import Exists, OuterRef, Q
 from django.utils.deprecation import RemovedInDjango31Warning
 
 UserModel = get_user_model()
@@ -119,6 +120,35 @@ def has_module_perms(self, user_obj, app_label):
             for perm in self.get_all_permissions(user_obj)
         )
 
+    def with_perm(self, perm, is_active=True, include_superusers=True, obj=None):
+        if not isinstance(perm, (str, Permission)):
+            raise TypeError('The `perm` argument must be a string or a permission instance.')
+        elif isinstance(perm, str):
+            try:
+                app_label, codename = perm.split('.')
+            except ValueError:
+                raise ValueError("Permission name should be in the form 'app_label.perm_name'.") from None
+
+        UserModel = get_user_model()
+
+        if obj is not None:
+            return UserModel._default_manager.none()
+
+        q0 = Q(group__user=OuterRef('pk')) | Q(user=OuterRef('pk'))
+        if isinstance(perm, Permission):
+            q0 &= Q(pk=perm.pk)
+        else:
+            q0 &= Q(codename=codename, content_type__app_label=app_label)
+
+        q1 = Q(has_permission=True)
+        if include_superusers:
+            q1 |= Q(is_superuser=True)
+        if is_active is not None:
+            q1 &= Q(is_active=is_active)
+
+        has_permission = Exists(Permission.objects.filter(q0))
+        return UserModel._default_manager.annotate(has_permission=has_permission).filter(q1)
+
     def get_user(self, user_id):
         try:
             user = UserModel._default_manager.get(pk=user_id)

django/contrib/auth/models.py

@@ -157,6 +157,22 @@ def create_superuser(self, username, email, password, **extra_fields):
 
         return self._create_user(username, email, password, **extra_fields)
 
+    def with_perm(self, perm, is_active=True, include_superusers=True, backend=None, obj=None):
+        if backend is None:
+            backends = auth._get_backends(return_tuples=True)
+            if len(backends) != 1:
+                raise ValueError(
+                    'You have multiple authentication backends configured and '
+                    'therefore must provide the `backend` argument.'
+                )
+            _, backend = backends[0]
+        elif not isinstance(backend, str):
+            raise TypeError('The `backend` argument must be a string.')
+        backend = auth.load_backend(backend)
+        if hasattr(backend, 'with_perm'):
+            return backend.with_perm(perm, is_active=is_active, include_superusers=include_superusers, obj=obj)
+        return self.get_queryset().none()
+
 
 # A few helper functions for common logic between User and AnonymousUser.
 def _user_get_permissions(user, obj, from_name):

docs/ref/contrib/auth.txt

@@ -287,6 +287,23 @@ Manager methods
         Same as :meth:`create_user`, but sets :attr:`~models.User.is_staff` and
         :attr:`~models.User.is_superuser` to ``True``.
 
+    .. method:: with_perm(perm, is_active=True, include_superusers=True, backend=None, obj=None)
+
+        .. versionadded:: 3.0
+
+        Returns a queryset containing :class:`~django.contrib.auth.models.User`
+        objects that have the given permission ``perm`` either in the form of
+        ``"<app label>.<permission codename>"`` or a
+        :class:`~django.contrib.auth.models.Permission` instance, including
+        superusers (if ``include_superusers`` is ``True``.)
+
+        If ``is_active`` is ``True``, only active users will be included, or if
+        ``False``, only inactive users will be included. Use ``None`` to return
+        all users irrespective of active state.
+
+        If ``backend`` is given, it will be used if it's defined in
+        the :setting:`AUTHENTICATION_BACKENDS` setting.
+
 ``AnonymousUser`` object
 ========================
 
@@ -516,6 +533,10 @@ The following backends are available in :mod:`django.contrib.auth.backends`:
     implement them other than returning an empty set of permissions if
     ``obj is not None``.
 
+    :meth:`with_perm` also allow an object to be passed as a parameter for
+    object-specific permissions, but it returns an empty queryset if
+    ``obj is not None``.
+
     .. method:: authenticate(request, username=None, password=None, **kwargs)
 
         Tries to authenticate ``username`` with ``password`` by calling
@@ -573,6 +594,15 @@ The following backends are available in :mod:`django.contrib.auth.backends`:
         don't have an :attr:`~django.contrib.auth.models.CustomUser.is_active`
         field are allowed.
 
+    .. method:: with_perm(perm, is_active=True, include_superusers=True, obj=None)
+
+        .. versionadded:: 3.0
+
+        Returns all active users who have the permission ``perm`` either in
+        the form of ``"<app label>.<permission codename>"`` or a
+        :class:`~django.contrib.auth.models.Permission` instance. Returns an
+        empty queryset if no users who have the ``perm`` found.
+
 .. class:: AllowAllUsersModelBackend
 
     Same as :class:`ModelBackend` except that it doesn't reject inactive users

docs/releases/3.0.txt

@@ -116,6 +116,10 @@ Minor features
   password and required fields, when a corresponding command line argument
   isn't provided in non-interactive mode.
 
+* The new
+  :meth:`UserManager.with_perm() <django.contrib.auth.models.UserManager.with_perm>`
+  method returns all users that have the specified permission.
+
 :mod:`django.contrib.contenttypes`
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

docs/topics/auth/customizing.txt

@@ -183,8 +183,9 @@ The user model will delegate permission lookup functions
 (:meth:`~django.contrib.auth.models.User.get_user_permissions()`,
 :meth:`~django.contrib.auth.models.User.get_group_permissions()`,
 :meth:`~django.contrib.auth.models.User.get_all_permissions()`,
-:meth:`~django.contrib.auth.models.User.has_perm()`, and
-:meth:`~django.contrib.auth.models.User.has_module_perms()`) to any
+:meth:`~django.contrib.auth.models.User.has_perm()`,
+:meth:`~django.contrib.auth.models.User.has_module_perms()`, and
+:meth:`~django.contrib.auth.models.UserManager.with_perm()`) to any
 authentication backend that implements these functions.
 
 The permissions given to the user will be the superset of all permissions

tests/auth_tests/models/__init__.py

@@ -4,7 +4,7 @@
 )
 from .invalid_models import CustomUserNonUniqueUsername
 from .is_active import IsActiveTestUser1
-from .minimal import MinimalUser
+from .minimal import CustomModel, MinimalUser
 from .no_password import NoPasswordUser
 from .proxy import Proxy, UserProxy
 from .uuid_pk import UUIDUser
@@ -13,6 +13,7 @@
 from .with_last_login_attr import UserWithDisabledLastLoginField
 
 __all__ = (
+    'CustomModel',
     'CustomPermissionsUser', 'CustomUser', 'CustomUserNonUniqueUsername',
     'CustomUserWithFK', 'CustomUserWithoutIsActiveField', 'Email',
     'ExtensionUser', 'IntegerUsernameUser', 'IsActiveTestUser1', 'MinimalUser',

tests/auth_tests/models/minimal.py

@@ -1,6 +1,12 @@
+from django.contrib.auth.models import User
 from django.db import models
 
 
 class MinimalUser(models.Model):
     REQUIRED_FIELDS = ()
     USERNAME_FIELD = 'id'
+
+
+class CustomModel(models.Model):
+    # Used by with_perm() tests.
+    user = models.ForeignKey(User, on_delete=models.CASCADE)

tests/auth_tests/test_auth_backends.py

@@ -696,7 +696,10 @@ class ImportedModelBackend(ModelBackend):
 
 
 class CustomModelBackend(ModelBackend):
-    pass
+    def with_perm(self, perm, is_active=True, include_superusers=True, backend=None, obj=None):
+        if obj is not None and obj.user.username == 'charliebrown':
+            return User.objects.filter(username='charliebrown')
+        return User.objects.filter(username__startswith='charlie')
 
 
 class OtherModelBackend(ModelBackend):

tests/auth_tests/test_models.py

@@ -12,7 +12,7 @@
 from django.db.models.signals import post_save
 from django.test import SimpleTestCase, TestCase, override_settings
 
-from .models import IntegerUsernameUser
+from .models import CustomModel, IntegerUsernameUser
 from .models.with_custom_email_field import CustomEmailField
 
 
@@ -261,6 +261,168 @@ def test_check_password_upgrade(self):
             hasher.iterations = old_iterations
 
 
+class UserWithPermTestCase(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.content_type = ContentType.objects.get_for_model(Group)
+        cls.permission = Permission.objects.create(
+            name='test', content_type=cls.content_type, codename='test'
+        )
+
+        cls.user1 = User.objects.create_user('user1', 'foo@example.com')
+        cls.user1.user_permissions.add(cls.permission)
+
+        cls.group = Group.objects.create(name='test')
+        cls.group.permissions.add(cls.permission)
+        cls.group2 = Group.objects.create(name='test2')
+        cls.group2.permissions.add(cls.permission)
+        cls.user2 = User.objects.create_user('user2', 'bar@example.com')
+        cls.user2.groups.add(cls.group, cls.group2)
+
+        # This user should not be found in the tests below.
+        cls.user3 = User.objects.create_user('user3', 'baz@example.com')
+
+        cls.superuser = User.objects.create_superuser(
+            'superuser', 'superuser@example.com', 'superpassword',
+        )
+        cls.inactive_user = User.objects.create_user(
+            'inactive_user', 'baz@example.com', is_active=False,
+        )
+        cls.inactive_user.user_permissions.add(cls.permission)
+        cls.charlie = User.objects.create_user('charlie', 'charlie@example.com')
+        cls.charlie_brown = User.objects.create_user('charliebrown', 'charlie@brown.com')
+
+    def test_valid_permission(self):
+        for perm in ('auth.test', self.permission):
+            with self.subTest(perm):
+                self.assertCountEqual(
+                    User.objects.with_perm(perm),
+                    [self.user1, self.user2, self.superuser],
+                )
+
+    def test_invalid_permission(self):
+        self.assertCountEqual(
+            User.objects.with_perm('invalid.perm'),
+            [self.superuser],
+        )
+
+    def test_include_superusers(self):
+        self.assertCountEqual(
+            User.objects.with_perm('auth.test', include_superusers=True),
+            [self.user1, self.user2, self.superuser],
+        )
+
+    def test_exclude_superusers(self):
+        self.assertCountEqual(
+            User.objects.with_perm('auth.test', include_superusers=False),
+            [self.user1, self.user2],
+        )
+
+    def test_active_users_only(self):
+        self.assertCountEqual(
+            User.objects.with_perm('auth.test', is_active=True),
+            [self.user1, self.user2, self.superuser],
+        )
+
+    def test_inactive_users_only(self):
+        self.assertCountEqual(
+            User.objects.with_perm('auth.test', is_active=False),
+            [self.inactive_user],
+        )
+
+    def test_active_and_inactive_users(self):
+        self.assertCountEqual(
+            User.objects.with_perm('auth.test', is_active=None),
+            [self.inactive_user, self.user1, self.user2, self.superuser],
+        )
+
+    def test_default_backend_with_permission_instance_and_obj(self):
+        obj = CustomModel.objects.create(user=self.charlie_brown)
+        self.assertCountEqual(
+            User.objects.with_perm(self.permission, obj=obj),
+            [],
+        )
+
+    @override_settings(AUTHENTICATION_BACKENDS=['auth_tests.test_auth_backends.CustomModelBackend'])
+    def test_custom_backend_with_permission_instance_and_obj(self):
+        obj = CustomModel.objects.create(user=self.charlie_brown)
+        self.assertCountEqual(
+            User.objects.with_perm(self.permission, obj=obj),
+            [self.charlie_brown],
+        )
+
+    def test_invalid_permission_name(self):
+        msg = "Permission name should be in the form 'app_label.perm_name'."
+        for perm in ('nodots', 'too.many.dots', '...', ''):
+            with self.subTest(perm=perm), self.assertRaisesMessage(ValueError, msg):
+                User.objects.with_perm(perm)
+
+    def test_invalid_permission_type(self):
+        class FakePermission:
+            pass
+
+        msg = 'The `perm` argument must be a string or a permission instance.'
+        for perm in (b'auth.test', FakePermission()):
+            with self.subTest(perm=perm), self.assertRaisesMessage(TypeError, msg):
+                User.objects.with_perm(perm)
+
+    @override_settings(AUTHENTICATION_BACKENDS=[
+        'auth_tests.test_auth_backends.CustomModelBackend',
+        'django.contrib.auth.backends.ModelBackend',
+    ])
+    def test_custom_backend(self):
+        msg = (
+            'You have multiple authentication backends configured and '
+            'therefore must provide the `backend` argument.'
+        )
+        backend = 'auth_tests.test_auth_backends.CustomModelBackend'
+        with self.assertRaisesMessage(ValueError, msg):
+            User.objects.with_perm('auth.test')
+        self.assertCountEqual(
+            User.objects.with_perm('auth.test', backend=backend),
+            [self.charlie, self.charlie_brown],
+        )
+        self.assertNotIn(self.user1, User.objects.with_perm('auth.test', backend=backend))
+
+    def test_default_backend_with_obj(self):
+        obj = CustomModel.objects.create(user=self.charlie_brown)
+        self.assertCountEqual(
+            User.objects.with_perm('auth.test', obj=obj),
+            [],
+        )
+
+    @override_settings(AUTHENTICATION_BACKENDS=['auth_tests.test_auth_backends.CustomModelBackend'])
+    def test_custom_backend_with_obj(self):
+        obj = CustomModel.objects.create(user=self.charlie_brown)
+        self.assertCountEqual(
+            User.objects.with_perm('auth.test', obj=obj),
+            [self.charlie_brown],
+        )
+
+    def test_invalid_backend_type(self):
+        msg = 'The `backend` argument must be a string.'
+        with self.assertRaisesMessage(TypeError, msg):
+            User.objects.with_perm(
+                'auth.test',
+                backend=b'auth_tests.test_auth_backends.CustomModelBackend',
+            )
+
+    def test_nonexistent_backend(self):
+        with self.assertRaises(ImportError):
+            User.objects.with_perm(
+                'auth.test',
+                backend='invalid.backend.CustomModelBackend',
+            )
+
+    @override_settings(AUTHENTICATION_BACKENDS=['django.contrib.auth.backends.BaseBackend'])
+    def test_backend_without_with_perm(self):
+        self.assertCountEqual(
+            User.objects.with_perm('auth.test'),
+            [],
+        )
+
+
 class IsActiveTestCase(TestCase):
     """
     Tests the behavior of the guaranteed is_active attribute