Merge branch 'release/2.025.37'

Author: kobo-bot[bot]

kobo/apps/accounts/apps.py

@@ -6,3 +6,9 @@
 class AccountExtrasConfig(AppConfig):
     name = 'kobo.apps.accounts'
     verbose_name = 'Account Extras'
+
+    def ready(self):
+        # Makes sure all signal handlers are connected
+        from kobo.apps.accounts import signals
+
+        super().ready()

kobo/apps/accounts/serializers.py

@@ -28,10 +28,18 @@ def create(self, validated_data):
     def validate(self, attrs):
         """
         Validates that only owners or admins of the organization can update
-        their email
+        their email and only if they don't have an SSO-provided email
         """
-        user = self.context['request'].user
+        request = self.context['request']
+        user = request.user
         organization = user.organization
+        # check if we have an SSO-provided email
+        # assume if we have an email address and an SSO account then the email comes
+        # from the SSO
+        if user.socialaccount_set.exists() and user.emailaddress_set.exists():
+            raise serializers.ValidationError(
+                {'email': t('This action is not allowed.')}
+            )
         if organization.is_owner(user) or organization.is_admin(user):
             return attrs
         raise serializers.ValidationError(
@@ -63,11 +71,11 @@ class Meta:
     @extend_schema_field(OpenApiTypes.EMAIL)
     def get_email(self, obj):
         if obj.extra_data:
-            if "email" in obj.extra_data:
-                return obj.extra_data.get("email")
-            return obj.extra_data.get("userPrincipalName")  # MS oauth uses this
+            if 'email' in obj.extra_data:
+                return obj.extra_data.get('email')
+            return obj.extra_data.get('userPrincipalName')  # MS oauth uses this
 
     @extend_schema_field(OpenApiTypes.STR)
     def get_username(self, obj):
         if obj.extra_data:
-            return obj.extra_data.get("username")
+            return obj.extra_data.get('username')

kobo/apps/accounts/signals.py

@@ -0,0 +1,44 @@
+from allauth.account.models import EmailAddress
+from allauth.account.signals import email_confirmed
+from allauth.account.utils import cleanup_email_addresses
+from allauth.socialaccount.signals import social_account_added
+from django.dispatch import receiver
+
+
+@receiver(social_account_added)
+def update_email(*args, **kwargs):
+    sociallogin = kwargs.get('sociallogin')
+    request = kwargs.get('request')
+    social_email_addresses = sociallogin.email_addresses
+    # if the provider doesn't use email, don't bother updating addresses
+    if not social_email_addresses:
+        return
+    social_user = sociallogin.user
+    for social_email in social_email_addresses:
+        # the auto-created EmailAddresses don't have the user already attached (?!)
+        social_email.user = social_user
+    existing_email_addresses = list(EmailAddress.objects.filter(user=social_user))
+    # put the social email addresses first so they get marked as primary
+    all_email_addresses = [*social_email_addresses, *existing_email_addresses]
+    emails, primary = cleanup_email_addresses(request, all_email_addresses)
+
+    # cleanup_email_addresses doesn't actually call set_as_primary on the primary
+    # email so do that now
+    primary.set_as_primary()
+
+    # update existing emails to reflect that they are no longer primary
+    # and add any new emails from the SocialLogin
+    EmailAddress.objects.bulk_create(
+        emails,
+        update_conflicts=True,
+        unique_fields=['email','user_id'],
+        update_fields=['primary']
+    )
+    # for some reason allauth doesn't emit the email confirmed signal even
+    # though if we're calling the social_account_added signal, the email has been
+    # verified
+    email_confirmed.send(
+        sender=EmailAddress,
+        request=request,
+        email_address=primary,
+    )

kobo/apps/accounts/tests/test_email.py

@@ -1,3 +1,5 @@
+from allauth.account.models import EmailAddress
+from ddt import data, ddt
 from django.conf import settings
 from django.core import mail
 from django.urls import reverse
@@ -84,7 +86,7 @@ def test_new_confirm_email(self):
                 confirm_url = line.split('testserver')[1].rsplit('/', 1)[0]
         queries = FuzzyInt(15, 20)
         with self.assertNumQueries(queries):
-            res = self.client.post(confirm_url + "/")
+            res = self.client.post(confirm_url + '/')
         self.assertEqual(res.status_code, 302)
         self.assertTrue(
             self.user.emailaddress_set.filter(
@@ -101,6 +103,7 @@ def test_new_confirm_email(self):
         )
 
 
+@ddt
 class EmailUpdateRestrictionTestCase(APITestCase):
     """
     Test that only organization owners and admins can update their email.
@@ -171,3 +174,28 @@ def test_that_non_mmo_user_can_update_email(self):
             ).count(),
             1
         )
+
+    @data('mmo_admin', 'mmo_owner', 'mmo_member', 'non_mmo_user')
+    def test_that_user_cannot_update_email_if_sso(self, user_type):
+        if user_type == 'mmo_admin':
+            user = self.admin
+        elif user_type == 'mmo_owner':
+            user = self.owner
+        elif user_type == 'mmo_member':
+            user = self.member
+        else:
+            user = self.non_mmo_user
+        baker.make('socialaccount.SocialAccount', user=user)
+        # in real life connecting the social account would have made an EmailAddress
+        email_address = baker.make('account.emailaddress', user=user)
+        self.assertNotEqual(email_address.email, '[email protected]')
+        self.client.force_login(user)
+        data = {'email': '[email protected]'}
+        res = self.client.post(self.url_list, data, format='json')
+        self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST)
+        # just check that the EmailAddress objects haven't changed;
+        # user.email relies on signal handlers so is tested in test_signals
+        self.assertEqual(user.emailaddress_set.count(), 1)
+        user_email = EmailAddress.objects.get(user=user)
+        self.assertEqual(user_email.email, email_address.email)
+        self.assertEqual(user.emailaddress_set.count(), 1)

kobo/apps/accounts/tests/test_signals.py

@@ -0,0 +1,35 @@
+from allauth.account.models import EmailAddress
+from allauth.socialaccount.models import SocialLogin
+from django.test import RequestFactory, TestCase
+
+from kobo.apps.accounts.signals import update_email
+from kobo.apps.kobo_auth.shortcuts import User
+
+
+class TestAccountSignals(TestCase):
+    fixtures = ['test_data']
+    def setUp(self):
+        self.user = User.objects.get(username='someuser')
+        self.email_address = EmailAddress.objects.create(
+            user=self.user,
+            email=self.user.email,
+            verified=True,
+            primary=True
+        )
+
+    def test_social_login_connect_updates_primary_email(self):
+        request = RequestFactory().get('/')
+        new_address = EmailAddress(
+            email='[email protected]',
+            verified=True,
+            primary=True
+        )
+        social_login = SocialLogin(email_addresses=[new_address], user=self.user)
+        update_email(request=request, sociallogin=social_login)
+        # we should have gotten rid of any old EmailAddresses
+        assert EmailAddress.objects.count() == 1
+        found_address = EmailAddress.objects.first()
+        assert found_address.email == new_address.email
+        assert found_address.verified
+        assert found_address.primary
+        assert self.user.email == new_address.email