[ENG-8749] Implement SSO attributes de-duplication for email and names

api/institutions/authentication.py

@@ -16,7 +16,8 @@
 from api.waffle.utils import flag_is_active
 
 from framework import sentry
-from framework.auth import get_or_create_institutional_user
+from framework.auth import get_or_create_institutional_user, deduplicate_sso_attributes
+from framework.auth.exceptions import MultipleSSOEmailError
 
 from osf import features
 from osf.exceptions import InstitutionAffiliationStateError
@@ -223,10 +224,16 @@ def authenticate(self, request):
                     f'sso_email={sso_email}, sso_identity={sso_identity}]',
                 )
 
+        # Deduplicate names first if it is provided
+        if fullname:
+            fullname = deduplicate_sso_attributes('fullname', fullname)
+        if given_name:
+            given_name = deduplicate_sso_attributes('given_name', given_name)
+        if family_name:
+            family_name = deduplicate_sso_attributes('family_name', family_name)
         # Use given name and family name to build full name if it is not provided
         if given_name and family_name and not fullname:
             fullname = given_name + ' ' + family_name
-
         # Non-empty full name is required. Fail the auth and inform sentry if not provided.
         if not fullname:
             message = f'Institution SSO Error: missing full name ' \
@@ -235,6 +242,14 @@ def authenticate(self, request):
             sentry.log_message(message)
             raise PermissionDenied(detail='InstitutionSsoMissingUserNames')
 
+        # Deduplicate sso email, currently we only handle duplicate sso email instead of multiple sso email
+        try:
+            sso_email = deduplicate_sso_attributes('sso_email', sso_email)
+        except MultipleSSOEmailError:
+            message = f'Institution SSO Error: multiple SSO email [sso_email={sso_email}, sso_identity={sso_identity}, institution={institution._id}]'
+            sentry.log_message(message)
+            logger.error(message)
+            raise PermissionDenied(detail='InstitutionSsoMultipleEmailsNotSupported')
         # Attempt to find an existing user that matches the email(s) provided via SSO. Create a new one if not found.
         # If a user is found, it is possible that the user is inactive (e.g. unclaimed, disabled, unconfirmed, etc.).
         # If a new user is created, the user object is confirmed but not registered (i.e. inactive until registered).

api_tests/institutions/views/test_institution_auth.py

@@ -494,6 +494,135 @@ def test_user_external_unconfirmed(self, app, institution, url_auth_institution)
         assert accepted_terms_of_service == user.accepted_terms_of_service
         assert not user.has_usable_password()
 
+    def test_duplicate_emails_and_names_success_existing_user(self, app, institution, url_auth_institution):
+        username, fullname, password = 'user_deanseu@user.edu', 'Foo Bar', 'FuAsKeEr'
+        exiting_user = make_user(username, fullname)
+        exiting_user.set_password(password)
+        exiting_user.save()
+        sso_email = f'{username};{username}'
+        with capture_signals() as mock_signals:
+            res = app.post(
+                url_auth_institution,
+                make_payload(
+                    institution,
+                    sso_email,
+                    family_name='User;User',
+                    given_name='Fake;Fake',
+                    fullname='Fake User;Fake User',
+                )
+            )
+        assert res.status_code == 204
+        assert not mock_signals.signals_sent()
+        user = OSFUser.objects.filter(username=username).first()
+        assert user
+        assert user.fullname == fullname
+        affiliation = user.get_institution_affiliation(institution._id)
+        assert affiliation.sso_mail == username
+        assert user.has_usable_password()
+        assert user.check_password(password)
+        assert institution in user.get_affiliated_institutions()
+
+    def test_duplicate_emails_and_names_success_new_user(self, app, institution, url_auth_institution):
+        username, fullname, family_name, given_name = 'user_deansnu@user.edu', 'Fake User', 'User', 'Fake'
+        sso_email = f'{username};{username}'
+        with capture_signals() as mock_signals:
+            res = app.post(
+                url_auth_institution,
+                make_payload(
+                    institution,
+                    sso_email,
+                    family_name=f'{family_name};{family_name}',
+                    given_name=f'{given_name};{given_name}',
+                    fullname=f'{fullname};{fullname}',
+                )
+            )
+        assert res.status_code == 204
+        assert mock_signals.signals_sent()
+        user = OSFUser.objects.filter(username=username).first()
+        assert user
+        assert user.fullname == fullname
+        assert user.family_name == family_name
+        assert user.given_name == given_name
+        affiliation = user.get_institution_affiliation(institution._id)
+        assert affiliation.sso_mail == username
+        assert user.has_usable_password()
+        assert institution in user.get_affiliated_institutions()
+
+    def test_multiple_names_warning_exiting_user(self, app, institution, url_auth_institution):
+        username, fullname, password = 'user_mnweu@user.edu', 'Foo Bar', 'FuAsKeEr'
+        exiting_user = make_user(username, fullname)
+        exiting_user.set_password(password)
+        exiting_user.save()
+        with capture_signals() as mock_signals:
+            res = app.post(
+                url_auth_institution,
+                make_payload(
+                    institution,
+                    username,
+                    family_name='User1;User2',
+                    given_name='Fake1;Fake2',
+                    fullname='Fake1 User1;Fake2 User2',
+                )
+            )
+        assert res.status_code == 204
+        assert not mock_signals.signals_sent()
+        user = OSFUser.objects.filter(username=username).first()
+        assert user
+        assert user.fullname == fullname
+        affiliation = user.get_institution_affiliation(institution._id)
+        assert affiliation.sso_mail == username
+        assert user.has_usable_password()
+        assert user.check_password(password)
+        assert institution in user.get_affiliated_institutions()
+
+    def test_multiple_names_warning_new_user(self, app, institution, url_auth_institution):
+        sso_email, fullname, family_name, given_name = 'user_deansnu@user.edu', 'Fake User;Foo Bar', 'User;Bar', 'Fake;Foo'
+        with capture_signals() as mock_signals:
+            res = app.post(
+                url_auth_institution,
+                make_payload(institution, sso_email, family_name=family_name, given_name=given_name, fullname=fullname),
+            )
+        assert res.status_code == 204
+        assert mock_signals.signals_sent()
+        user = OSFUser.objects.filter(username=sso_email).first()
+        assert user
+        assert user.fullname == fullname
+        assert user.family_name == family_name
+        assert user.given_name == given_name
+        affiliation = user.get_institution_affiliation(institution._id)
+        assert affiliation.sso_mail == sso_email
+        assert user.has_usable_password()
+        assert institution in user.get_affiliated_institutions()
+
+    def test_multiple_emails_failure_existing_user(self, app, institution, url_auth_institution):
+        username, second_email, fullname, password = 'user_mefeu_a', 'user_mefeu_b@user.edu', 'Fake User', 'FuAsKeEr'
+        existing_uesr = make_user(username, fullname)
+        existing_uesr.set_password(password)
+        existing_uesr.save()
+        sso_email = f'{username};{second_email}'
+        with capture_signals() as mock_signals:
+            res = app.post(
+                url_auth_institution,
+                make_payload(institution, sso_email=sso_email, fullname=fullname),
+                expect_errors=True,
+            )
+        assert res.status_code == 403
+        assert res.json['errors'][0]['detail'] == 'InstitutionSsoMultipleEmailsNotSupported'
+        assert not mock_signals.signals_sent()
+
+    def test_multiple_emails_failure_new_user(self, app, institution, url_auth_institution):
+        first_email, second_email, family_name, given_name = 'user_mefeu_a', 'user_mefeu_b@user.edu', 'User', 'Fake'
+        sso_email = f'{first_email};{second_email}'
+        with capture_signals() as mock_signals:
+            res = app.post(
+                url_auth_institution,
+                make_payload(institution, sso_email, family_name=family_name, given_name=given_name),
+                expect_errors=True,
+            )
+        assert res.status_code == 403
+        assert res.json['errors'][0]['detail'] == 'InstitutionSsoMultipleEmailsNotSupported'
+        assert not mock_signals.signals_sent()
+
 
 @pytest.mark.django_db
 class TestInstitutionStorageRegion:

framework/auth/__init__.py

@@ -2,11 +2,11 @@
 
 from django.utils import timezone
 
-from framework import bcrypt
+from framework import bcrypt, sentry
 from framework.auth import signals
 from framework.auth.core import Auth
 from framework.auth.core import get_user, generate_verification_key
-from framework.auth.exceptions import DuplicateEmailError
+from framework.auth.exceptions import DuplicateEmailError, MultipleSSOEmailError
 from framework.auth.tasks import update_user_from_activity
 from framework.auth.utils import LogLevel, print_cas_log
 from framework.celery_tasks.handlers import enqueue_task
@@ -159,6 +159,19 @@ def get_or_create_institutional_user(fullname, sso_email, sso_identity, primary_
     return user, True, None, None, sso_identity
 
 
+def deduplicate_sso_attributes(attr_name, attr_value, delimiter=';'):
+    if delimiter not in attr_value:
+        return attr_value
+    value_set = set(attr_value.split(delimiter))
+    if len(value_set) != 1:
+        message = f'Multiple values found for SSO attribute: [{attr_name}={attr_value}]'
+        sentry.log_message(message)
+        if attr_name == 'sso_email':
+            raise MultipleSSOEmailError(message)
+        return attr_value
+    return value_set.pop()
+
+
 def get_or_create_user(fullname, address, reset_password=True, is_spam=False):
     """
     Get or create user by fullname and email address.

framework/auth/exceptions.py

@@ -65,3 +65,8 @@ class MergeConflictError(EmailConfirmTokenError):
     """Raised if a merge is not possible due to a conflict"""
     message_short = language.CANNOT_MERGE_ACCOUNTS_SHORT
     message_long = language.CANNOT_MERGE_ACCOUNTS_LONG
+
+
+class MultipleSSOEmailError(AuthError):
+    """Raised if institution SSO provides multiple emails which OSF cannot deduplicate."""
+    pass