OpenConceptLab/ocl_online#482 | SSO sign-in matches a legacy mixed-case username on its verified email instead of creating a case twin (#958)

Author: snyaggarwal

core/common/backends.py

@@ -8,7 +8,8 @@
 from pydash import get
 from redis import Sentinel
 
-from core.common.exceptions import DeactivatedAccountLoginRefused, SSOIdentityMismatchRefused
+from core.common.exceptions import DeactivatedAccountLoginRefused, SSOIdentityMismatchRefused, SSOSignInRefused, \
+    SSOUsernameCaseMismatchRefused
 
 logger = logging.getLogger('oclapi')
 
@@ -97,19 +98,16 @@ def update_user(self, user, claims):
     def authenticate(self, request, **kwargs):
         try:
             return super().authenticate(request, **kwargs)
-        except (DeactivatedAccountLoginRefused, SSOIdentityMismatchRefused):
+        except SSOSignInRefused:
             # the refused identity's tokens were stored in the session before its account was looked up
             for key in ('oidc_access_token', 'oidc_id_token'):
                 request.session.pop(key, None)
             return None
 
     def filter_users_by_claims(self, claims):
         """
-        Matches the account linked to the claims' Keycloak id (`sub`), else the one with the claims' username exactly
-        as Keycloak sends it, unless that account is linked to a different Keycloak user. A deactivated account is never
-        matched on its username alone (ocl_online#339): it's reactivated when the claims carry its email, verified by
-        Keycloak, and otherwise the sign-in is refused and the account left untouched. The username can't go to a new
-        account either, since usernames are unique.
+        Matches the account linked to the claims' `sub`, else by username exactly, else ignoring case. A deactivated
+        or case-only match needs the claims' verified email; otherwise the sign-in is refused, never a new account.
         """
         from core.users.models import UserProfile
 
@@ -123,6 +121,8 @@ def filter_users_by_claims(self, claims):
         users = UserProfile.objects.filter(sso_id=sub) if sub else UserProfile.objects.none()
         if not users:
             users = UserProfile.objects.filter(username=username)
+            if not users:
+                users = self.filter_users_by_username_ignoring_case(username, claims)
             if users and sub and users[0].sso_id not in (None, sub):
                 logger.warning('OIDC sign-in refused: user %s is linked to another Keycloak user', users[0].id)
                 raise SSOIdentityMismatchRefused()
@@ -137,6 +137,19 @@ def filter_users_by_claims(self, claims):
 
         return users
 
+    def filter_users_by_username_ignoring_case(self, username, claims):
+        from core.users.models import UserProfile
+
+        users = UserProfile.objects.filter(username__iexact=username)
+        if not users:
+            return users
+        if len(users) > 1 or not self.has_verified_email_of(users[0], claims):
+            logger.warning(
+                'OIDC sign-in refused: users %s match the username only ignoring case, and the claims lack the '
+                'verified email of exactly one', [user.id for user in users])
+            raise SSOUsernameCaseMismatchRefused()
+        return users
+
     @staticmethod
     def has_verified_email_of(user, claims):
         email = claims.get('email')

core/common/exceptions.py

@@ -28,7 +28,11 @@ class Http403(APIException):
     default_code = 'forbidden'
 
 
-class DeactivatedAccountLoginRefused(AuthenticationFailed):
+class SSOSignInRefused(AuthenticationFailed):
+    pass
+
+
+class DeactivatedAccountLoginRefused(SSOSignInRefused):
     """
     Raised when a Keycloak sign-in's username matches a deactivated account but the claims don't carry that
     account's email, verified by Keycloak (ocl_online#339). The account is left as it was; staff can reactivate it.
@@ -42,7 +46,7 @@ def __init__(self):
         )
 
 
-class SSOIdentityMismatchRefused(AuthenticationFailed):
+class SSOIdentityMismatchRefused(SSOSignInRefused):
     """Raised when a Keycloak sign-in's username matches an account already linked to a different Keycloak user."""
     default_code = 'sso_identity_mismatch'
 
@@ -53,6 +57,17 @@ def __init__(self):
         )
 
 
+class SSOUsernameCaseMismatchRefused(SSOSignInRefused):
+    """Raised when a Keycloak sign-in's username matches an account only ignoring case, without its verified email."""
+    default_code = 'sso_username_case_mismatch'
+
+    def __init__(self):
+        super().__init__(
+            'This username belongs to an existing OCL account with different capitalization. If it is yours, '
+            f'contact {settings.COMMUNITY_EMAIL} and we will link it to your sign-in.'
+        )
+
+
 class BatchIndexingError(Exception):
     """
     Raised once a batched ES indexing run has attempted every batch and one or more failed (ocl_online#241).

core/common/tests.py

@@ -76,7 +76,7 @@
 from core.users.models import UserProfile
 from core.users.tests.factories import UserProfileFactory
 from .backends import OCLOIDCAuthenticationBackend
-from .exceptions import SSOIdentityMismatchRefused
+from .exceptions import SSOIdentityMismatchRefused, SSOUsernameCaseMismatchRefused
 from .checksums import Checksum
 from .serializers import IdentifierSerializer
 from .validators import URIValidator
@@ -2802,20 +2802,71 @@ def test_update_user_keeps_a_stored_sso_id(self):
         self.backend.update_user(user, {**self.claim, 'sub': 'sub-2'})
         self.assertEqual(user.sso_id, 'sub-1')
 
-    def test_filter_users_by_claims_matches_username_exactly(self):
-        """Keycloak sends usernames in lowercase, and a mixed-case account is never matched, active or not"""
+    def test_filter_users_by_claims_prefers_the_exact_username(self):
         UserProfileFactory(username='Batman', email='batman@gotham.com')
+        batman = UserProfileFactory(username='batman', email='batman@gotham.com')
+
+        with self.assertNumQueries(1):
+            users = self.backend.filter_users_by_claims(self.claim)
+            self.assertEqual(list(users), [batman])
+
+    def test_filter_users_by_claims_matches_username_ignoring_case_with_its_verified_email(self):
+        batman = UserProfileFactory(username='Batman', email=' Batman@Gotham.com')
+
+        users = self.backend.filter_users_by_claims({**self.claim, 'sub': 'sub-1'})
+        self.assertEqual(list(users), [batman])
+        self.backend.update_user(users[0], {**self.claim, 'sub': 'sub-1'})
+
+        batman.refresh_from_db()
+        self.assertEqual(batman.username, 'Batman')
+        self.assertEqual(batman.sso_id, 'sub-1')
+        self.assertEqual(UserProfile.objects.filter(username__iexact='batman').count(), 1)
+
+    def test_filter_users_by_claims_refuses_username_ignoring_case_without_its_verified_email(self):
+        batman = UserProfileFactory(username='Batman', email='batman@gotham.com', first_name='Bat')
+        without_email_verified = {key: value for key, value in self.claim.items() if key != 'email_verified'}
+
+        for claims in [
+                {**self.claim, 'email_verified': False},
+                {**self.claim, 'email_verified': 'true'},
+                without_email_verified,
+                {**self.claim, 'email': 'joker@gotham.com'},
+                {**self.claim, 'email': None},
+                {**self.claim, 'email': ''},
+        ]:
+            with self.subTest(claims=claims):
+                with patch('core.users.models.UserProfile.save') as save_mock:
+                    with self.assertRaises(SSOUsernameCaseMismatchRefused):
+                        self.backend.filter_users_by_claims(claims)
+                save_mock.assert_not_called()
+
+        batman.refresh_from_db()
+        self.assertEqual(batman.first_name, 'Bat')
+        self.assertIsNone(batman.sso_id)
+
+    def test_filter_users_by_claims_refuses_several_accounts_matching_username_ignoring_case(self):
+        UserProfileFactory(username='Batman', email='batman@gotham.com')
+        UserProfileFactory(username='BatMan', email='joker@gotham.com')
+
+        with self.assertRaises(SSOUsernameCaseMismatchRefused):
+            self.backend.filter_users_by_claims(self.claim)
+
+    def test_filter_users_by_claims_refuses_username_ignoring_case_linked_to_a_different_sub(self):
+        UserProfileFactory(username='Batman', email='batman@gotham.com', sso_id='sub-1')
+
+        with self.assertRaises(SSOIdentityMismatchRefused):
+            self.backend.filter_users_by_claims({**self.claim, 'sub': 'sub-2'})
+
+    def test_filter_users_by_claims_reactivates_deactivated_user_ignoring_case_with_its_verified_email(self):
         robin = UserProfileFactory(username='Robin', email='robin@gotham.com')
         robin.deactivate()
 
-        self.assertEqual(self.backend.filter_users_by_claims(self.claim).count(), 0)
-        self.assertEqual(
-            self.backend.filter_users_by_claims(
-                {**self.claim, 'preferred_username': 'robin', 'email': 'robin@gotham.com'}).count(),
-            0
-        )
+        users = self.backend.filter_users_by_claims(
+            {**self.claim, 'preferred_username': 'robin', 'email': 'robin@gotham.com'})
+
+        self.assertEqual(list(users), [robin])
         robin.refresh_from_db()
-        self.assertFalse(robin.is_active)
+        self.assertTrue(robin.is_active)
 
     def test_filter_users_by_claims_reactivates_deactivated_user_with_its_verified_email(self):
         batman = UserProfileFactory(username='batman', email='Batman@Gotham.com ')
@@ -2936,19 +2987,24 @@ def test_get_or_create_user_for_deactivated_user(self, get_userinfo_mock):
         self.assertEqual(batman.first_name, 'Bruce')
 
     @patch('core.common.backends.OCLOIDCAuthenticationBackend.get_userinfo')
-    def test_get_or_create_user_leaves_mixed_case_deactivated_user(self, get_userinfo_mock):
-        get_userinfo_mock.return_value = self.claim
+    def test_get_or_create_user_for_mixed_case_user(self, get_userinfo_mock):
         old_batman = UserProfileFactory(username='Batman', email='batman@gotham.com', first_name='Bat')
-        old_batman.deactivate()
 
+        get_userinfo_mock.return_value = {**self.claim, 'email': 'joker@gotham.com'}
+        with patch.object(self.backend, 'create_user') as create_user_mock:
+            with self.assertRaises(SSOUsernameCaseMismatchRefused):
+                self.backend.get_or_create_user('access-token', None, None)
+        create_user_mock.assert_not_called()
+        self.assertEqual(UserProfile.objects.filter(username__iexact='batman').count(), 1)
+
+        get_userinfo_mock.return_value = self.claim
         user = self.backend.get_or_create_user('access-token', None, None)
 
-        self.assertNotEqual(user, old_batman)
-        self.assertEqual(user.username, 'batman')
-        self.assertTrue(user.is_active)
+        self.assertEqual(user, old_batman)
+        self.assertEqual(UserProfile.objects.filter(username__iexact='batman').count(), 1)
         old_batman.refresh_from_db()
-        self.assertFalse(old_batman.is_active)
-        self.assertEqual(old_batman.first_name, 'Bat')
+        self.assertEqual(old_batman.username, 'Batman')
+        self.assertEqual(old_batman.first_name, 'Bruce')
 
     @patch('mozilla_django_oidc.auth.OIDCAuthenticationBackend.authenticate')
     def test_authenticate_fails_and_drops_stored_tokens_when_sign_in_is_refused(self, authenticate_mock):

core/middlewares/middlewares.py

@@ -14,7 +14,7 @@
 from core.common.authentication import OCLAuthentication
 from core.common.constants import VERSION_HEADER, REQUEST_USER_HEADER, RESPONSE_TIME_HEADER, REQUEST_URL_HEADER, \
     REQUEST_METHOD_HEADER
-from core.common.exceptions import DeactivatedAccountLoginRefused, SSOIdentityMismatchRefused
+from core.common.exceptions import SSOSignInRefused
 from core.common.throttling import FhirMinuteThrottle, ThrottleUtil
 from core.common.utils import set_current_user, set_request_url
 from core.fhir.constants import FHIR_JSON
@@ -135,7 +135,7 @@ def __call__(self, request):
         """Allow exempt and approved anonymous traffic, otherwise return 403."""
         try:
             is_allowed = self.is_request_allowed(request)
-        except (DeactivatedAccountLoginRefused, SSOIdentityMismatchRefused) as ex:
+        except SSOSignInRefused as ex:
             # a refused sign-in says why, rather than reading as anonymous access (ocl_online#339)
             headers = {'WWW-Authenticate': 'Bearer realm="api"'}
             if is_fhir_path(request.path):
@@ -181,7 +181,7 @@ def get_authenticated_user(request):
 
         try:
             auth_result = OCLAuthentication().authenticate(Request(request))
-        except (DeactivatedAccountLoginRefused, SSOIdentityMismatchRefused):
+        except SSOSignInRefused:
             raise
         except AuthenticationFailed:
             return user

core/middlewares/tests.py

@@ -7,7 +7,8 @@
 from django.test import RequestFactory, SimpleTestCase, override_settings
 from rest_framework.exceptions import AuthenticationFailed
 
-from core.common.exceptions import DeactivatedAccountLoginRefused, SSOIdentityMismatchRefused
+from core.common.exceptions import DeactivatedAccountLoginRefused, SSOIdentityMismatchRefused, \
+    SSOUsernameCaseMismatchRefused
 from core.middlewares.middlewares import RequireAuthenticationMiddleware
 
 
@@ -93,6 +94,15 @@ def test_rejects_sign_in_refused_for_identity_mismatch(self, authenticate_mock):
         self.assertEqual(response['WWW-Authenticate'], 'Bearer realm="api"')
         self.assertIn('different sign-in identity', json.loads(response.content)['detail'])
 
+    @patch('core.middlewares.middlewares.OCLAuthentication.authenticate')
+    def test_rejects_sign_in_refused_for_username_case_mismatch(self, authenticate_mock):
+        authenticate_mock.side_effect = SSOUsernameCaseMismatchRefused()
+
+        response = self.middleware(self.make_request('/user/', HTTP_AUTHORIZATION='Bearer sso-token'))
+
+        self.assertEqual(response.status_code, 401)
+        self.assertIn('different capitalization', json.loads(response.content)['detail'])
+
     def test_blocks_anonymous_request_for_protected_path(self):
         """Anonymous traffic to protected API paths should receive a 403 response."""
         response = self.middleware(self.make_request('/orgs/OCL/'))