[#1672] Reworked auth backends, contacts to support non-unique (digid) emails

src/open_inwoner/accounts/backends.py

@@ -26,14 +26,28 @@ def authenticate(
         self, request, username=None, password=None, user=None, token=None, **kwargs
     ):
         config = SiteConfiguration.get_solo()
-
+        User = get_user_model()
         if username and password and not config.login_2fa_sms:
             try:
-                user = get_user_model().objects.get(email__iexact=username)
-                if check_password(password, user.password):
+                user = User.objects.get(
+                    email__iexact=username,
+                    login_type=LoginTypeChoices.default,
+                )
+                if check_password(
+                    password, user.password
+                ) and self.user_can_authenticate(user):
                     return user
-            except get_user_model().DoesNotExist:
+            except User.MultipleObjectsReturned:
+                # Found multiple users with this email (shouldn't happen if we added checks)
+                # Run the default password hasher once to reduce the timing
+                # difference between an existing and a nonexistent user (#20760).
+                User().set_password(password)
+                return None
+            except User.DoesNotExist:
                 # No user was found, return None - triggers default login failed
+                # Run the default password hasher once to reduce the timing
+                # difference between an existing and a nonexistent user (#20760).
+                User().set_password(password)
                 return None
 
         # 2FA with sms verification
@@ -57,18 +71,29 @@ def authenticate(self, request=None, *args, **kwargs):
 
 class CustomOIDCBackend(OIDCAuthenticationBackend):
     def create_user(self, claims):
-        """Return object for a newly created user account."""
+        """
+        Return object for a newly created user account.
+
+        before we got here we already checked for existing users based on the overriden queryset from the .filter_users_by_claims()
+        """
         unique_id = self.retrieve_identifier_claim(claims)
 
-        email = generate_email_from_string(unique_id)
         if "email" in claims:
             email = claims["email"]
+        else:
+            email = generate_email_from_string(unique_id)
 
-        existing_user = self.UserModel.objects.filter(email__iexact=email).first()
+        existing_user = self.UserModel.objects.filter(
+            email__iexact=email,
+            login_type=LoginTypeChoices.default,
+            is_active=True,
+        ).first()
         if existing_user:
             logger.debug("Updating OIDC user: %s with email %s", unique_id, email)
             existing_user.oidc_id = unique_id
             existing_user.login_type = LoginTypeChoices.oidc
+            # TODO verify we want unusable_password
+            existing_user.set_unusable_password()
             existing_user.save()
             return existing_user
         else:
@@ -79,9 +104,10 @@ def create_user(self, claims):
                 "email": email,
                 "login_type": LoginTypeChoices.oidc,
             }
-
             user = self.UserModel.objects.create_user(**kwargs)
             self.update_user(user, claims)
+            # TODO verify we want unusable_password
+            user.set_unusable_password()
 
             return user
 

src/open_inwoner/accounts/forms.py

@@ -211,6 +211,11 @@ def __init__(self, user, *args, **kwargs):
 
 
 class CustomPasswordResetForm(PasswordResetForm):
+    def get_users(self, email):
+        users = super().get_users(email)
+        # filter regular email login users
+        return [u for u in users if u.login_type == LoginTypeChoices.default]
+
     def send_mail(
         self,
         subject_template_name,
@@ -222,28 +227,26 @@ def send_mail(
     ):
         """
         Send a django.core.mail.EmailMultiAlternatives to `to_email`.
+
+        Note: the context has the user specific information / tokens etc
         """
-        email = self.cleaned_data.get("email")
-        user = User.objects.get(email=email)
-
-        if user.login_type == LoginTypeChoices.default:
-            subject = loader.render_to_string(subject_template_name, context)
-            # Email subject *must not* contain newlines
-            subject = "".join(subject.splitlines())
-            body = loader.render_to_string(email_template_name, context)
-
-            email_message = EmailMultiAlternatives(
-                subject,
-                body,
-                from_email,
-                [to_email],
-                headers={"X-Mail-Queue-Priority": "now"},
-            )
-            if html_email_template_name is not None:
-                html_email = loader.render_to_string(html_email_template_name, context)
-                email_message.attach_alternative(html_email, "text/html")
+        subject = loader.render_to_string(subject_template_name, context)
+        # Email subject *must not* contain newlines
+        subject = "".join(subject.splitlines())
+        body = loader.render_to_string(email_template_name, context)
+
+        email_message = EmailMultiAlternatives(
+            subject,
+            body,
+            from_email,
+            [to_email],
+            headers={"X-Mail-Queue-Priority": "now"},
+        )
+        if html_email_template_name is not None:
+            html_email = loader.render_to_string(html_email_template_name, context)
+            email_message.attach_alternative(html_email, "text/html")
 
-            email_message.send()
+        email_message.send()
 
 
 class CategoriesForm(forms.ModelForm):
@@ -302,73 +305,63 @@ def __init__(self, user, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
     def clean(self):
+        """
+        Note cleaning and lookup is a bit convoluted as we have to deal with non-unique emails:
+         - adding multiple contacts at same time
+         - users adding their own email, to add their other account as contact
+
+        But we still want to provide some error feedback
+        """
         cleaned_data = super().clean()
-        first_name = cleaned_data.get("first_name")
-        last_name = cleaned_data.get("last_name")
         email = cleaned_data.get("email")
-
-        try:
-            contact_user = self.find_user(email)
-        except (ValidationError, User.MultipleObjectsReturned):
-            contact_user = self.find_user(email, first_name, last_name)
-
-        cleaned_data["contact_user"] = contact_user
-
-    def find_user(
-        self,
-        email: str,
-        first_name: Optional[str] = None,
-        last_name: Optional[str] = None,
-    ):
-        """Try to find a user by email alone, or email + first_name + last_name"""
-
-        existing_users = User.objects.filter(email__iexact=email)
-
-        # no user found: return, then send invitation (no ValidationError raised)
-        if not existing_users:
+        if not email:
             return
 
-        if first_name and last_name:
-            existing_users = User.objects.filter(
-                Q(first_name__iexact=first_name) & Q(last_name__iexact=last_name)
-            )
+        # use sets for simplicity, and use .only("id")
+        existing_users = set(User.objects.filter(email__iexact=email))
+        user_contacts = set(self.user.user_contacts.all().only("id"))
+        contacts_for_approval = set(self.user.contacts_for_approval.all().only("id"))
 
-        existing_users = existing_users.filter(is_active=True)
+        # check if this was our own email and if we just found ourselves
+        if self.user in existing_users:
+            existing_users.remove(self.user)
+            if not existing_users:
+                raise ValidationError(_("You cannot add yourself as a contact."))
 
-        # no active users with the given specs
         if not existing_users:
-            raise ValidationError(
-                _("The user cannot be added, their account has been deleted.")
-            )
+            # no users found, pass and let the view send an Invite to the email
+            return
 
-        # check if there are multiple users with the given specs
-        try:
-            existing_user = existing_users.get()
-        except User.MultipleObjectsReturned:
-            raise ValidationError(
-                _(
-                    "We're having trouble finding an account with this information."
-                    "Please contact us for help."
+        # best effort, we're going to return successful if we find at least one good contact
+        #   or only report the worst error (to not confuse the end-user)
+        not_active = False
+        has_contact = False
+        added_contacts = set()
+
+        # check if these users are valid and not already added
+        for contact_user in existing_users:
+            if not contact_user.is_active:
+                not_active = True
+            elif contact_user in user_contacts or contact_user in contacts_for_approval:
+                has_contact = True
+            else:
+                added_contacts.add(contact_user)
+
+        # remember the contacts and let the view add records, logs and the emails
+        if added_contacts:
+            cleaned_data["added_contacts"] = added_contacts
+        else:
+            # produce some feedback, check most interesting first
+            if has_contact:
+                raise ValidationError(
+                    _(
+                        "Het ingevoerde e-mailadres komt al voor in uw contactpersonen. Pas de gegevens aan en probeer het opnieuw."
+                    )
                 )
-            )
-
-        # contact already exists
-        if self.user.has_contact(existing_user):
-            raise ValidationError(
-                _(
-                    "Het ingevoerde contact komt al voor in uw contactpersonen. "
-                    "Pas de gegevens aan en probeer het opnieuw."
+            elif not_active:
+                raise ValidationError(
+                    _("The user cannot be added, their account has been deleted.")
                 )
-            )
-
-        # user attempts to add themselves as contact
-        if (
-            self.user.first_name == existing_user.first_name
-            and self.user.last_name == existing_user.last_name
-        ):
-            raise ValidationError(_("You cannot add yourself as a contact."))
-
-        return existing_user
 
 
 class UserField(forms.ModelChoiceField):

src/open_inwoner/accounts/managers.py

@@ -33,6 +33,8 @@ def get_by_rsin(self, rsin):
         return self.get_queryset().get(rsin=rsin)
 
     def eherkenning_create(self, rsin, **kwargs):
+        raise NotImplementedError("old code, please verify before use")
+        # TODO what is this @rsin.com email hostname? @example.org is bad enough but this actually exists
         return super().create(
             email="user-{}@rsin.com".format(rsin),
             login_type=LoginTypeChoices.eherkenning,

src/open_inwoner/accounts/migrations/0065_user_unique_email_when_not_digid.py

@@ -0,0 +1,21 @@
+# Generated by Django 3.2.20 on 2023-09-07 09:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("accounts", "0064_alter_user_email"),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name="user",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(("login_type", "digid"), _negated=True),
+                fields=("email",),
+                name="unique_email_when_not_digid",
+            ),
+        ),
+    ]