Extra renaming considerations and tests for conflicts and false positives

django_openid_auth/auth.py

@@ -81,7 +81,7 @@ class OpenIDBackend:
 
         if getattr(settings, 'OPENID_UPDATE_DETAILS_FROM_SREG', False):
             details = self._extract_user_details(openid_response)
-            self.update_user_details(user, details)
+            self.update_user_details(user, details, openid_response)
 
         teams_response = teams.TeamsResponse.fromSuccessResponse(
             openid_response)
@@ -98,7 +98,6 @@ class OpenIDBackend:
             email = sreg_response.get('email')
             fullname = sreg_response.get('fullname')
             nickname = sreg_response.get('nickname')
-
         # If any attributes are provided via Attribute Exchange, use
         # them in preference.
         fetch_response = ax.FetchResponse.fromSuccessResponse(openid_response)
@@ -137,10 +136,36 @@ class OpenIDBackend:
         return dict(email=email, nickname=nickname,
                     first_name=first_name, last_name=last_name)
 
-    def create_user_from_openid(self, openid_response):
-        details = self._extract_user_details(openid_response)
-        nickname = details['nickname'] or 'openiduser'
-        email = details['email'] or ''
+    def _get_available_username(self, nickname, identity_url):
+        nickname = nickname or 'openiduser'
+        # See if we already have this nickname assigned to a username
+        try:
+            user = User.objects.get(username__exact=nickname)
+        except User.DoesNotExist:
+            # No conflict, we can use this nickname
+            return nickname
+            
+        # Check if we already have nickname+i for this identity_url
+        try:
+            user_openid = UserOpenID.objects.get(
+                claimed_id__exact=identity_url,
+                user__username__startswith=nickname)
+            # No exception means we have an existing user for this identity
+            # that starts with this nickname, so it's possible we've had to
+            # assign them to nickname+i already.
+            oid_username = user_openid.user.username
+            if len(oid_username) > len(nickname):
+                try:
+                    # check that it ends with a number
+                    int(oid_username[len(nickname):])
+                    return oid_username
+                except ValueError:
+                    # username starts with nickname, but isn't nickname+#
+                    pass
+        except UserOpenID.DoesNotExist:
+            # No user associated with this identity_url
+            pass
+            
 
         # Pick a username for the user based on their nickname,
         # checking for conflicts.
@@ -150,15 +175,27 @@ class OpenIDBackend:
             if i > 1:
                 username += str(i)
             try:
-                User.objects.get(username__exact=username)
+                user = User.objects.get(username__exact=username)
+                if user.useropenid_set.filter(claimed_id__exact=identity_url).count() > 0:
+                    # username already belongs to this openid user, so it's okay
+                    return username
+                    
             except User.DoesNotExist:
                 break
             i += 1
+        return username
+    
+    def create_user_from_openid(self, openid_response):
+        details = self._extract_user_details(openid_response)
+        nickname = details['nickname'] or 'openiduser'
+        email = details['email'] or ''
 
-        user = User.objects.create_user(username, email, password=None)
-        self.update_user_details(user, details)
+        username = self._get_available_username(details['nickname'], openid_response.identity_url)
 
+        user = User.objects.create_user(username, email, password=None)
         self.associate_openid(user, openid_response)
+        self.update_user_details(user, details, openid_response)
+
         return user
 
     def associate_openid(self, user, openid_response):
@@ -181,7 +218,7 @@ class OpenIDBackend:
 
         return user_openid
 
-    def update_user_details(self, user, details):
+    def update_user_details(self, user, details, openid_response):
         updated = False
         if details['first_name']:
             user.first_name = details['first_name']
@@ -193,7 +230,7 @@ class OpenIDBackend:
             user.email = details['email']
             updated = True
         if getattr(settings, 'OPENID_FOLLOW_RENAMES', False):
-            user.username = details['nickname']
+            user.username = self._get_available_username(details['nickname'], openid_response.identity_url)
             updated = True
 
         if updated:

django_openid_auth/tests/test_views.py

@@ -135,12 +135,14 @@ class RelyingPartyTests(TestCase):
         self.old_sso_server_url = getattr(settings, 'OPENID_SSO_SERVER_URL', None)
         self.old_teams_map = getattr(settings, 'OPENID_LAUNCHPAD_TEAMS_MAPPING', {})
         self.old_use_as_admin_login = getattr(settings, 'OPENID_USE_AS_ADMIN_LOGIN', False)
+        self.old_follow_renames = getattr(settings, 'OPENID_FOLLOW_RENAMES', False)
 
         settings.OPENID_CREATE_USERS = False
         settings.OPENID_UPDATE_DETAILS_FROM_SREG = False
         settings.OPENID_SSO_SERVER_URL = None
         settings.OPENID_LAUNCHPAD_TEAMS_MAPPING = {}
         settings.OPENID_USE_AS_ADMIN_LOGIN = False
+        settings.OPENID_FOLLOW_RENAMES = False
 
     def tearDown(self):
         settings.LOGIN_REDIRECT_URL = self.old_login_redirect_url
@@ -149,6 +151,7 @@ class RelyingPartyTests(TestCase):
         settings.OPENID_SSO_SERVER_URL = self.old_sso_server_url
         settings.OPENID_LAUNCHPAD_TEAMS_MAPPING = self.old_teams_map
         settings.OPENID_USE_AS_ADMIN_LOGIN = self.old_use_as_admin_login
+        settings.OPENID_FOLLOW_RENAMES = self.old_follow_renames
 
         setDefaultFetcher(None)
         super(RelyingPartyTests, self).tearDown()
@@ -286,23 +289,9 @@ class RelyingPartyTests(TestCase):
         self.assertEquals(user.last_name, 'User')
         self.assertEquals(user.email, 'foo@example.com')
 
-    def test_login_update_details_rename(self):
-        settings.OPENID_FOLLOW_RENAMES = True
-        return self.test_login_update_details()
-        
-    def test_login_update_details(self):
-        settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
-        user = User.objects.create_user('testuser', 'someone@example.com')
-        useropenid = UserOpenID(
-            user=user,
-            claimed_id='http://example.com/identity',
-            display_id='http://example.com/identity')
-        useropenid.save()
-
+    def _do_user_login(self, openid_req, openid_resp):
         # Posting in an identity URL begins the authentication request:
-        response = self.client.post('/openid/login/',
-            {'openid_identifier': 'http://example.com/identity',
-             'next': '/getuser/'})
+        response = self.client.post('/openid/login/', openid_req)
         self.assertContains(response, 'OpenID transaction in progress')
 
         # Complete the request, passing back some simple registration
@@ -311,22 +300,180 @@ class RelyingPartyTests(TestCase):
         sreg_request = sreg.SRegRequest.fromOpenIDRequest(openid_request)
         openid_response = openid_request.answer(True)
         sreg_response = sreg.SRegResponse.extractResponse(
-            sreg_request, {'nickname': 'someuser', 'fullname': 'Some User',
-                           'email': 'foo@example.com'})
+            sreg_request, openid_resp)
         openid_response.addExtension(sreg_response)
         response = self.complete(openid_response)
+#        self.assertEqual(openid_resp, (response.status_code, response.content))
         self.assertRedirects(response, 'http://testserver/getuser/')
 
+    def test_login_without_nickname(self):
+        settings.OPENID_CREATE_USERS = True
+
+        openid_req = {'openid_identifier': 'http://example.com/identity',
+               'next': '/getuser/'}
+        openid_resp =  {'nickname': '', 'fullname': 'Openid User',
+                 'email': 'foo@example.com'}
+        self._do_user_login(openid_req, openid_resp)
+        response = self.client.get('/getuser/')
+
+        # username defaults to 'openiduser'
+        self.assertEquals(response.content, 'openiduser')
+
+        # The user's full name and email have been updated.
+        user = User.objects.get(username=response.content)
+        self.assertEquals(user.first_name, 'Openid')
+        self.assertEquals(user.last_name, 'User')
+        self.assertEquals(user.email, 'foo@example.com')
+        
+    def test_login_follow_rename(self):
+        settings.OPENID_FOLLOW_RENAMES = True
+        settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
+        user = User.objects.create_user('testuser', 'someone@example.com')
+        useropenid = UserOpenID(
+            user=user,
+            claimed_id='http://example.com/identity',
+            display_id='http://example.com/identity')
+        useropenid.save()
+
+        openid_req = {'openid_identifier': 'http://example.com/identity',
+               'next': '/getuser/'}
+        openid_resp =  {'nickname': 'someuser', 'fullname': 'Some User',
+                 'email': 'foo@example.com'}
+        self._do_user_login(openid_req, openid_resp)
+        response = self.client.get('/getuser/')
+
         # If OPENID_FOLLOW_RENAMES, they are logged in as 
         # someuser (the passed in nickname has changed the username)
-        #
-        # Otherwise they are now logged in as testuser (the passed in
-        # nickname has not caused the username to change).
+        self.assertEquals(response.content, 'someuser')
+
+        # The user's full name and email have been updated.
+        user = User.objects.get(username=response.content)
+        self.assertEquals(user.first_name, 'Some')
+        self.assertEquals(user.last_name, 'User')
+        self.assertEquals(user.email, 'foo@example.com')
+        
+    def test_login_follow_rename_conflict(self):
+        settings.OPENID_FOLLOW_RENAMES = True
+        settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
+        # Setup existing user who's name we're going to switch to
+        user = User.objects.create_user('testuser', 'someone@example.com')
+        UserOpenID.objects.get_or_create(
+            user=user,
+            claimed_id='http://example.com/existing_identity',
+            display_id='http://example.com/existing_identity')
+
+        # Setup user who is going to try to change username to 'testuser'
+        renamed_user = User.objects.create_user('renameuser', 'someone@example.com')
+        UserOpenID.objects.get_or_create(
+            user=renamed_user,
+            claimed_id='http://example.com/identity',
+            display_id='http://example.com/identity')
+
+        # identity url is for 'renameuser'
+        openid_req = {'openid_identifier': 'http://example.com/identity',
+               'next': '/getuser/'}
+        # but returned username is for 'testuser', which already exists for another identity
+        openid_resp =  {'nickname': 'testuser', 'fullname': 'Rename User',
+                 'email': 'rename@example.com'}
+        self._do_user_login(openid_req, openid_resp)
         response = self.client.get('/getuser/')
-        if getattr(settings, 'OPENID_FOLLOW_RENAMES', False):
-            self.assertEquals(response.content, 'someuser')
-        else:
-            self.assertEquals(response.content, 'testuser')
+
+        # If OPENID_FOLLOW_RENAMES, attempt to change username to 'testuser'
+        # but since that username is already taken by someone else, we go through
+        # the process of adding +i to it, and get testuser2.
+        self.assertEquals(response.content, 'testuser2')
+
+        # The user's full name and email have been updated.
+        user = User.objects.get(username=response.content)
+        self.assertEquals(user.first_name, 'Rename')
+        self.assertEquals(user.last_name, 'User')
+        self.assertEquals(user.email, 'rename@example.com')
+        
+    def test_login_follow_rename_conflict_onlyonce(self):
+        settings.OPENID_FOLLOW_RENAMES = True
+        settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
+        # Setup existing user who's name we're going to switch to
+        user = User.objects.create_user('testuser', 'someone@example.com')
+        UserOpenID.objects.get_or_create(
+            user=user,
+            claimed_id='http://example.com/existing_identity',
+            display_id='http://example.com/existing_identity')
+
+        # Setup user who is going to try to change username to 'testuser'
+        renamed_user = User.objects.create_user('testuser2000', 'someone@example.com')
+        UserOpenID.objects.get_or_create(
+            user=renamed_user,
+            claimed_id='http://example.com/identity',
+            display_id='http://example.com/identity')
+
+        # identity url is for 'testuser2000'
+        openid_req = {'openid_identifier': 'http://example.com/identity',
+               'next': '/getuser/'}
+        # but returned username is for 'testuser', which already exists for another identity
+        openid_resp =  {'nickname': 'testuser', 'fullname': 'Rename User',
+                 'email': 'rename@example.com'}
+        self._do_user_login(openid_req, openid_resp)
+        response = self.client.get('/getuser/')
+
+        # If OPENID_FOLLOW_RENAMES, attempt to change username to 'testuser'
+        # but since that username is already taken by someone else, we go through
+        # the process of adding +i to it.  Since the user for this identity url
+        # already has a name matching that pattern, check if first.
+        self.assertEquals(response.content, 'testuser2000')
+
+        # The user's full name and email have been updated.
+        user = User.objects.get(username=response.content)
+        self.assertEquals(user.first_name, 'Rename')
+        self.assertEquals(user.last_name, 'User')
+        self.assertEquals(user.email, 'rename@example.com')
+        
+    def test_login_follow_rename_false_conflict(self):
+        settings.OPENID_FOLLOW_RENAMES = True
+        settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
+        # Setup existing user who's username matches the name+i pattern
+        user = User.objects.create_user('testuser2', 'someone@example.com')
+        UserOpenID.objects.get_or_create(
+            user=user,
+            claimed_id='http://example.com/identity',
+            display_id='http://example.com/identity')
+
+        # identity url is for 'testuser2'
+        openid_req = {'openid_identifier': 'http://example.com/identity',
+               'next': '/getuser/'}
+        # but returned username is for 'testuser', which looks like we've done
+        # a username+1 for them already, but 'testuser' isn't actually taken
+        openid_resp =  {'nickname': 'testuser', 'fullname': 'Same User',
+                 'email': 'same@example.com'}
+        self._do_user_login(openid_req, openid_resp)
+        response = self.client.get('/getuser/')
+
+        # If OPENID_FOLLOW_RENAMES, username should be changed to 'testuser'
+        # because it wasn't currently taken
+        self.assertEquals(response.content, 'testuser')
+
+        # The user's full name and email have been updated.
+        user = User.objects.get(username=response.content)
+        self.assertEquals(user.first_name, 'Same')
+        self.assertEquals(user.last_name, 'User')
+        self.assertEquals(user.email, 'same@example.com')
+        
+    def test_login_update_details(self):
+        settings.OPENID_UPDATE_DETAILS_FROM_SREG = True
+        user = User.objects.create_user('testuser', 'someone@example.com')
+        useropenid = UserOpenID(
+            user=user,
+            claimed_id='http://example.com/identity',
+            display_id='http://example.com/identity')
+        useropenid.save()
+
+        openid_req = {'openid_identifier': 'http://example.com/identity',
+               'next': '/getuser/'}
+        openid_resp =  {'nickname': 'testuser', 'fullname': 'Some User',
+                 'email': 'foo@example.com'}
+        self._do_user_login(openid_req, openid_resp)
+        response = self.client.get('/getuser/')
+
+        self.assertEquals(response.content, 'testuser')
 
         # The user's full name and email have been updated.
         user = User.objects.get(username=response.content)