Implement duplicate user protection

account_manager/models.py

@@ -12,6 +12,8 @@ from ldapdb.models.base import Model
 from core.settings import PASSWORD_RESET_TIMEOUT_DAYS
 from account_manager.utils.mail_utils import realm_send_mail
 from multiprocessing import Process
+from ldap import NO_SUCH_OBJECT, ALREADY_EXISTS
+from django.core.exceptions import ObjectDoesNotExist
 
 
 class LdapUser(Model):
@@ -42,22 +44,26 @@ class LdapUser(Model):
 
     @staticmethod
     def create_with_django_user_creation_and_welcome_mail(realm, protocol, domain, username, email):
-        ldap_user = LdapUser.objects.create(username=username, email=email, first_name=" ", last_name=" ")
-        user, _ = User.objects.get_or_create(username=username, email=email)
-        mail_subject = 'Activate your blog account.'
-        message = render_to_string('registration/welcome_email.jinja2', {
-            'user': user,
-            'domain': domain,
-            'uid': urlsafe_base64_encode(force_bytes(user.pk)).decode(),
-            'token': default_token_generator.make_token(user=user),
-            'protocol': protocol,
-            'email': email,
-            'expiration_days': PASSWORD_RESET_TIMEOUT_DAYS
-        })
-        # TODO failure handling
-        p1 = Process(target=realm_send_mail, args=(realm, user.email, mail_subject, message))
-        p1.start()
-        return ldap_user
+        if not LdapUser.is_user_duplicate(username):
+            LdapUser.base_dn = f'ou=people, {realm.ldap_base_dn}'
+            ldap_user = LdapUser.objects.create(username=username, email=email, first_name=" ", last_name=" ")
+            user, _ = User.objects.get_or_create(username=username, email=email)
+            mail_subject = 'Activate your blog account.'
+            message = render_to_string('registration/welcome_email.jinja2', {
+                'user': user,
+                'domain': domain,
+                'uid': urlsafe_base64_encode(force_bytes(user.pk)).decode(),
+                'token': default_token_generator.make_token(user=user),
+                'protocol': protocol,
+                'email': email,
+                'expiration_days': PASSWORD_RESET_TIMEOUT_DAYS
+            })
+            # TODO failure handling
+            p1 = Process(target=realm_send_mail, args=(realm, user.email, mail_subject, message))
+            p1.start()
+            return ldap_user
+        else:
+            raise ALREADY_EXISTS('User already exists')
 
     @staticmethod
     def password_reset(user, raw_password):
@@ -67,6 +73,15 @@ class LdapUser(Model):
         LdapUser.base_dn = re.compile('(uid=[a-zA-Z0-9_]*),(.*)').match(ldap_user.dn).group(2)
         ldap_user.save()
 
+    @staticmethod
+    def is_user_duplicate(username):
+        LdapUser.base_dn = LdapUser.ROOT_DN
+        try:
+            LdapUser.objects.get(username=username)
+            return True
+        except (NO_SUCH_OBJECT, ObjectDoesNotExist) as err:
+            return False
+
 
 class LdapGroup(Model):
     """