Merge pull request #240 from jordiromera/ldap_clean

Ldap clean users

Author: etianen

README.rst

@@ -86,7 +86,7 @@ Available settings
 
     # The LDAP username and password of a user for querying the LDAP database for user
     # details. If None, then the authenticated user will be used for querying, and
-    # the `ldap_sync_users` command will perform an anonymous query.
+    # the `ldap_sync_users`, `ldap_clean_users` commands will perform an anonymous query.
     LDAP_AUTH_CONNECTION_USERNAME = None
     LDAP_AUTH_CONNECTION_PASSWORD = None
 
@@ -158,6 +158,19 @@ The parameters are:-
 - ``dn`` - the DN (Distinguished Name) of the LDAP matched user (optional keyword only parameter)
 
 
+Clean User
+----------
+
+When a LDAP user is removed from server it could be interresting to deactive or delete its local Django account
+to prevent unauthorized access.
+
+To do so run:
+
+    ``./manage.py ldap_clean_users`` (or ``./manage.py ldap_clean_users --purge``).
+
+It will deactivate all local users non declared on LDAP server. If ``--purge`` is specified, all local users will be deleted.
+
+
 Can't get authentication to work?
 ---------------------------------
 

django_python3_ldap/ldap.py

@@ -116,17 +116,27 @@ def get_user(self, **kwargs):
         in settings.LDAP_AUTH_USER_LOOKUP_FIELDS.
         """
         # Search the LDAP database.
-        if self._connection.search(
+        if self.has_user(**kwargs):
+            return self._get_or_create_user(self._connection.response[0])
+        logger.warning("LDAP user lookup failed")
+        return None
+
+    def has_user(self, **kwargs):
+        """
+        Returns True if the user with the given identifier exists.
+
+        The user identifier should be keyword arguments matching the fields
+        in settings.LDAP_AUTH_USER_LOOKUP_FIELDS.
+        """
+        # Search the LDAP database.
+        return self._connection.search(
             search_base=settings.LDAP_AUTH_SEARCH_BASE,
             search_filter=format_search_filter(kwargs),
             search_scope=ldap3.SUBTREE,
             attributes=ldap3.ALL_ATTRIBUTES,
             get_operational_attributes=True,
             size_limit=1,
-        ):
-            return self._get_or_create_user(self._connection.response[0])
-        logger.warning("LDAP user lookup failed")
-        return None
+        )
 
 
 @contextmanager

django_python3_ldap/management/commands/ldap_clean_users.py

@@ -0,0 +1,112 @@
+from django.contrib.auth import get_user_model
+from django.core.management.base import BaseCommand, CommandError
+from django.db import transaction
+from django.db.models import ProtectedError
+
+from django_python3_ldap import ldap
+from django_python3_ldap.conf import settings
+from django_python3_ldap.utils import group_lookup_args
+
+
+class Command(BaseCommand):
+
+    help = "Remove local user models for users not find anymore in the remote LDAP authentication server."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '-p',
+            '--purge',
+            action='store_true',
+            help='Purge instead of deactive local user models'
+        )
+        parser.add_argument(
+            'lookups',
+            nargs='*',
+            type=str,
+            help='A list of lookup values, matching the fields specified in LDAP_AUTH_USER_LOOKUP_FIELDS. '
+                 'If this is not provided then ALL users are concerned.'
+        )
+        parser.add_argument(
+            '--superuser',
+            action='store_true',
+            help='Handle superuser (by default, superusers are excluded)'
+        )
+        parser.add_argument(
+            '--staff',
+            action='store_true',
+            help='Handle staff user (by default,staff users are excluded)'
+        )
+
+    @staticmethod
+    def _iter_local_users(User, lookups, superuser, staff):
+        """
+        Iterates over local users. If the list of lookups is empty, then all users are returned.
+        However, if lookups are provided, User.object.get is used to clean each user found using the lookups.
+        Exclude or not superuser and or staff user.
+        """
+
+        if len(lookups) < 1:
+            for user in User.objects.filter(is_superuser=superuser,
+                                            is_staff=staff):
+                yield user
+        else:
+            for lookup in group_lookup_args(*lookups):
+                try:
+                    yield User.objects.get(**lookup,
+                                           is_superuser=superuser,
+                                           is_staff=staff)
+                except User.DoesNotExist:
+                    raise CommandError("Could not find user with lookup : {lookup}".format(
+                        lookup=lookup,
+                    ))
+
+    @staticmethod
+    def _remove(user, purge):
+        """
+        Deactivate or purge a given local user
+        """
+        if purge:
+            # Delete local user
+            try:
+                user.delete()
+            except ProtectedError as e:
+                raise CommandError("Could not purge user {user} : {e}".format(
+                    user=user,
+                    e=e
+                ))
+        else:
+            # Deactivate local user
+            user.is_active = False
+            user.save()
+
+    @transaction.atomic()
+    def handle(self, *args, **kwargs):
+        verbosity = int(kwargs.get("verbosity", 1))
+        purge = kwargs.get('purge', False)
+        lookups = kwargs.get('lookups', [])
+        superuser = kwargs.get('superuser', False)
+        staff = kwargs.get('staff', False)
+        User = get_user_model()
+        auth_kwargs = {
+            User.USERNAME_FIELD: settings.LDAP_AUTH_CONNECTION_USERNAME,
+            'password': settings.LDAP_AUTH_CONNECTION_PASSWORD
+        }
+        with ldap.connection(**auth_kwargs) as connection:
+            if connection is None:
+                raise CommandError("Could not connect to LDAP server")
+            for user in self._iter_local_users(User, lookups, superuser, staff):
+                # For each local users
+                # Check if user still exists
+                user_kwargs = {
+                    User.USERNAME_FIELD: getattr(user, User.USERNAME_FIELD)
+                }
+                if connection.has_user(**user_kwargs):
+                    # User still exists on LDAP side
+                    continue
+                # Clean user
+                self._remove(user, purge)
+                if verbosity >= 1:
+                    self.stdout.write("{action} {user}".format(
+                        action=('Purged' if purge else 'Deactivated'),
+                        user=user,
+                    ))

django_python3_ldap/tests.py

@@ -57,6 +57,20 @@ def testGetUserKwargsIncorrectUsername(self):
             )
             self.assertEqual(user, None)
 
+    def testHasUserKwargsSuccess(self):
+        with connection() as c:
+            exist = c.has_user(
+                username=settings.LDAP_AUTH_TEST_USER_USERNAME,
+            )
+            self.assertEqual(exist, True)
+
+    def testHasUserKwargsIncorrectUsername(self):
+        with connection() as c:
+            exist = c.has_user(
+                username="bad" + settings.LDAP_AUTH_TEST_USER_USERNAME,
+            )
+            self.assertEqual(exist, False)
+
     # Authentication tests.
 
     def testAuthenticateUserSuccess(self):
@@ -247,3 +261,154 @@ def testImportFunc(self):
 
         with self.settings(LDAP_AUTH_SYNC_USER_RELATIONS='django.contrib.auth.get_user_model'):
             self.assertTrue(callable(import_func(settings.LDAP_AUTH_SYNC_USER_RELATIONS)))
+
+    def testCleanUsersDeactivate(self):
+        """
+        ldap_clean_users management command test
+        """
+        from django.contrib.auth import get_user_model
+        User = get_user_model()
+        _username = "nonldap{user}".format(user=settings.LDAP_AUTH_TEST_USER_USERNAME)
+        user = User.objects.create_user(
+            _username,
+            "nonldap{mail}".format(mail=settings.LDAP_AUTH_TEST_USER_EMAIL),
+            settings.LDAP_AUTH_TEST_USER_PASSWORD)
+        user.save()
+        user_count_1 = User.objects.count()
+        self.assertEqual(User.objects.get(username=_username).is_active, True)
+        call_command("ldap_clean_users", verbosity=0)
+        user_count_2 = User.objects.count()
+        self.assertEqual(user_count_1, user_count_2)
+        self.assertEqual(User.objects.get(username=_username).is_active, False)
+
+        """
+        Test with lookup
+        """
+        # Reactivate user
+        user = User.objects.get(username=_username)
+        user.is_active = True
+        user.save()
+        # Create second user
+        _usernameLookup = "nonldaplookup{user}".format(user=settings.LDAP_AUTH_TEST_USER_USERNAME)
+        user = User.objects.create_user(
+            _usernameLookup,
+            "nonldaplookup{mail}".format(mail=settings.LDAP_AUTH_TEST_USER_EMAIL),
+            settings.LDAP_AUTH_TEST_USER_PASSWORD)
+        user.save()
+        user_count_1 = User.objects.count()
+        self.assertEqual(User.objects.get(username=_usernameLookup).is_active, True)
+        # Clean second user
+        call_command("ldap_clean_users", _usernameLookup, verbosity=0)
+        user_count_2 = User.objects.count()
+        self.assertEqual(user_count_1, user_count_2)
+        self.assertEqual(User.objects.get(username=_usernameLookup).is_active, False)
+        self.assertEqual(User.objects.get(username=_username).is_active, True)
+        # Reactivate second user
+        user = User.objects.get(username=_usernameLookup)
+        user.is_active = True
+        user.save()
+        # Clean first user
+        call_command("ldap_clean_users", _username, verbosity=0)
+        self.assertEqual(User.objects.get(username=_username).is_active, False)
+        self.assertEqual(User.objects.get(username=_usernameLookup).is_active, True)
+        # Lookup a non existing user (raise a CommandError)
+        with self.assertRaises(CommandError):
+            call_command("ldap_clean_users", 'doesnonexist', verbosity=0)
+
+        """
+        Test with superuser
+        """
+        # Reactivate first user and promote to superuser
+        user = User.objects.get(username=_username)
+        user.is_active = True
+        user.is_superuser = True
+        user.save()
+        # Reactivate second user
+        user = User.objects.get(username=_usernameLookup)
+        user.is_active = True
+        user.save()
+        call_command("ldap_clean_users", superuser=False, verbosity=0)
+        self.assertEqual(User.objects.get(username=_username).is_active, True)
+        self.assertEqual(User.objects.get(username=_usernameLookup).is_active, False)
+        call_command("ldap_clean_users", superuser=True, verbosity=0)
+        self.assertEqual(User.objects.get(username=_username).is_active, False)
+
+        """
+        Test with staff user
+        """
+        # Reactivate first user and promote to staff
+        user = User.objects.get(username=_username)
+        user.is_active = True
+        user.is_superuser = False
+        user.is_staff = True
+        user.save()
+        # Reactivate second user
+        user = User.objects.get(username=_usernameLookup)
+        user.is_active = True
+        user.save()
+        call_command("ldap_clean_users", staff=False, verbosity=0)
+        self.assertEqual(User.objects.get(username=_username).is_active, True)
+        self.assertEqual(User.objects.get(username=_usernameLookup).is_active, False)
+        call_command("ldap_clean_users", staff=True, verbosity=0)
+        self.assertEqual(User.objects.get(username=_username).is_active, False)
+
+    def testCleanUsersPurge(self):
+        """
+        ldap_clean_users management command test with purge argument
+        """
+        from django.contrib.auth import get_user_model
+        User = get_user_model()
+        user = User.objects.create_user(
+            "nonldap{user}".format(user=settings.LDAP_AUTH_TEST_USER_USERNAME),
+            "nonldap{mail}".format(mail=settings.LDAP_AUTH_TEST_USER_EMAIL),
+            settings.LDAP_AUTH_TEST_USER_PASSWORD)
+        user.save()
+        user_count_1 = User.objects.count()
+        call_command("ldap_clean_users", verbosity=0, purge=True)
+        user_count_2 = User.objects.count()
+        self.assertGreater(user_count_1, user_count_2)
+
+    def testCleanUsersCommandOutput(self):
+        # Test without purge
+        out = StringIO()
+        from django.contrib.auth import get_user_model
+        User = get_user_model()
+        user = User.objects.create_user(
+            "nonldap{user}".format(user=settings.LDAP_AUTH_TEST_USER_USERNAME),
+            "nonldap{mail}".format(mail=settings.LDAP_AUTH_TEST_USER_EMAIL),
+            settings.LDAP_AUTH_TEST_USER_PASSWORD)
+        user.save()
+        call_command("ldap_clean_users", stdout=out, verbosity=1)
+        rows = out.getvalue().split("\n")[:-1]
+        self.assertEqual(len(rows), 1)
+        for row in rows:
+            self.assertRegex(row, r'^Deactivated ')
+        # Reset for next test
+        user.delete()
+        out.truncate(0)
+        out.seek(0)
+        # Test with purge
+        user = User.objects.create_user(
+            "nonldap{user}".format(user=settings.LDAP_AUTH_TEST_USER_USERNAME),
+            "nonldap{mail}".format(mail=settings.LDAP_AUTH_TEST_USER_EMAIL),
+            settings.LDAP_AUTH_TEST_USER_PASSWORD)
+        user.save()
+        call_command("ldap_clean_users", stdout=out, verbosity=1, purge=True)
+        rows = out.getvalue().split("\n")[:-1]
+        self.assertEqual(len(rows), 1)
+        for row in rows:
+            self.assertRegex(row, r'^Purged ')
+
+    def testReCleanUsersDoesntRecreateUsers(self):
+        from django.contrib.auth import get_user_model
+        User = get_user_model()
+        user = User.objects.create_user(
+            "nonldap{user}".format(user=settings.LDAP_AUTH_TEST_USER_USERNAME),
+            "nonldap{mail}".format(mail=settings.LDAP_AUTH_TEST_USER_EMAIL),
+            settings.LDAP_AUTH_TEST_USER_PASSWORD)
+        user.save()
+        call_command("ldap_clean_users", verbosity=0, purge=True)
+        user_count_1 = User.objects.count()
+        call_command("ldap_clean_users", verbosity=0, purge=True)
+        user_count_2 = User.objects.count()
+        self.assertEqual(user_count_1, user_count_2)