[fix] Simplified implementation

Author: pandafy

openwisp_radius/saml/utils.py

@@ -9,16 +9,3 @@ def get_url_or_path(url):
     if parsed_url.netloc:
         return f'{parsed_url.scheme}://{parsed_url.netloc}{parsed_url.path}'
     return parsed_url.path
-
-
-def get_email_from_ava(ava):
-    email_keys = (
-        'email',
-        'mail',
-        'uid',
-    )
-    for key in email_keys:
-        email = ava.get(key, None)
-        if email is not None:
-            return email[0]
-    return None

openwisp_radius/saml/views.py

@@ -2,9 +2,8 @@
 from urllib.parse import parse_qs, quote, urlencode, urlparse
 
 import swapper
-from allauth.account.models import EmailAddress
 from allauth.account.utils import send_email_confirmation
-from allauth.utils import ValidationError
+from allauth.utils import valid_email_or_none
 from django import forms
 from django.conf import settings
 from django.contrib.auth import get_user_model, logout
@@ -24,7 +23,7 @@
 from .. import settings as app_settings
 from ..api.views import RadiusTokenMixin
 from ..utils import get_organization_radius_settings, load_model
-from .utils import get_email_from_ava, get_url_or_path
+from .utils import get_url_or_path
 
 logger = logging.getLogger(__name__)
 
@@ -75,42 +74,20 @@ def post_login_hook(self, request, user, session_info):
         try:
             user.registered_user
         except ObjectDoesNotExist:
-            email = None
-            uid_is_email = 'email' in getattr(
-                settings, 'SAML_ATTRIBUTE_MAPPING', {}
-            ).get('uid', ())
-            if uid_is_email:
-                email = session_info['name_id'].text
-            if email is None:
-                email = get_email_from_ava(session_info['ava'])
-            if email:
-                user.email = email
-                try:
-                    user.full_clean()
-                    user.save()
-                    EmailAddress.objects.create(
-                        user=user, email=email, verified=True, primary=True
-                    )
-                except ValidationError:
-                    assertion_email = get_email_from_ava(session_info['ava'])
-                    if assertion_email and assertion_email != email:
-                        user.email = assertion_email
-                        try:
-                            user.full_clean()
-                            user.save()
-                            EmailAddress.objects.create(
-                                user=user,
-                                email=assertion_email,
-                                verified=True,
-                                primary=True,
-                            )
-                        except ValidationError:
-                            raise ValidationError('Email Verification Failed')
             registered_user = RegisteredUser(
                 user=user, method='saml', is_verified=app_settings.SAML_IS_VERIFIED
             )
             registered_user.full_clean()
             registered_user.save()
+            # The user is just created, it will not have an email address
+            if user.email:
+                email = valid_email_or_none(user.email)
+                if not email:
+                    logger.exception(
+                        f'Failed email validation for "{user}"'
+                        ' during SAML user creation'
+                    )
+            send_email_confirmation(request, user, signup=True, email=user.email)
 
     def customize_relay_state(self, relay_state):
         """

openwisp_radius/tests/test_saml/test_views.py

@@ -3,11 +3,9 @@
 from urllib.parse import parse_qs, urlparse
 
 import swapper
-from allauth.account.models import EmailAddress
 from django.conf import settings
 from django.contrib.auth import SESSION_KEY, get_user_model
 from django.core import mail
-from django.core.validators import ValidationError
 from django.test import TestCase, override_settings
 from django.urls import reverse, reverse_lazy
 from djangosaml2.tests import auth_response, conf
@@ -73,8 +71,7 @@ def _post_successful_auth_assertions(self, query_params, org_slug):
         self.assertEqual(User.objects.count(), 1)
         user_id = self.client.session[SESSION_KEY]
         user = User.objects.get(id=user_id)
-        email = EmailAddress.objects.filter(user=user)
-        self.assertEqual(email.count(), 1)
+        self.assertEqual(user.emailaddress_set.count(), 1)
         self.assertEqual(user.username, '[email protected]')
         self.assertEqual(OrganizationUser.objects.count(), 1)
         org_user = OrganizationUser.objects.get(user_id=user_id)
@@ -118,14 +115,18 @@ def test_invalid_email_raise_validation_error(self):
         saml_response, relay_state = self._get_saml_response_for_acs_view(
             relay_state, uid=invalid_email
         )
-        with self.assertRaises(ValidationError):
+        with patch('logging.Logger.exception') as mocked_logger:
             self.client.post(
                 reverse('radius:saml2_acs'),
                 {
                     'SAMLResponse': self.b64_for_post(saml_response),
                     'RelayState': relay_state,
                 },
             )
+        mocked_logger.assert_called_once_with(
+            'Failed email validation for "invalid_email@example" during'
+            ' SAML user creation'
+        )
 
     @capture_any_output()
     def test_relay_state_relative_path(self):