feat: improved user attribute handling and username uniqueness in SpidCieOidcRpCallbackView

vendor/spid-cie-oidc-django/spid_cie_oidc/relying_party/views/rp_callback.py

@@ -65,6 +65,25 @@ class SpidCieOidcRpCallbackView(View, SpidCieOidcRp, OidcUserInfo, OAuth2Authori
 
     def user_reunification(self, user_attrs: dict):
         user_model = get_user_model()
+        desired_username = user_attrs.get("username", user_attrs["sub"])
+
+        def update_user(user):
+            user.first_name = user_attrs.get("given_name", user.first_name)
+            user.last_name = user_attrs.get("family_name", user.last_name)
+            user.email = user_attrs.get("email", user.email)
+            merged_attrs = dict(user.attributes or {})
+            merged_attrs.update(user_attrs)
+            user.attributes = merged_attrs
+            user.save()
+            return user
+
+        def unique_username(base_username):
+            username = base_username
+            index = 1
+            while user_model.objects.filter(username=username).exists():
+                username = f"{base_username}__{index}"
+                index += 1
+            return username
 
         if user_attrs.get(RP_USER_LOOKUP_FIELD, None):
             lookup = {
@@ -77,13 +96,34 @@ class SpidCieOidcRpCallbackView(View, SpidCieOidcRp, OidcUserInfo, OAuth2Authori
             }
         user = user_model.objects.filter(**lookup).first()
         if user:
-            user.attributes.update(user_attrs)
-            user.save()
+            user = update_user(user)
             logger.info(f"{RP_USER_LOOKUP_FIELD} matched on user {user}")
             return user
         elif RP_USER_CREATE:
+            existing_username_user = user_model.objects.filter(
+                username=desired_username
+            ).first()
+            if existing_username_user:
+                existing_lookup_value = (existing_username_user.attributes or {}).get(
+                    RP_USER_LOOKUP_FIELD
+                )
+                requested_lookup_value = user_attrs.get(RP_USER_LOOKUP_FIELD)
+                if (
+                    not requested_lookup_value
+                    or not existing_lookup_value
+                    or existing_lookup_value == requested_lookup_value
+                ):
+                    user = update_user(existing_username_user)
+                    logger.info(f"Username matched on existing user {user}")
+                    return user
+                logger.warning(
+                    "Username collision for %s with different %s values, generating a unique username",
+                    desired_username,
+                    RP_USER_LOOKUP_FIELD,
+                )
+
             user = user_model.objects.create(
-                username=user_attrs.get("username", user_attrs["sub"]),
+                username=unique_username(desired_username),
                 first_name=user_attrs.get("given_name", user_attrs["sub"]),
                 last_name=user_attrs.get("family_name", user_attrs["sub"]),
                 email=user_attrs.get("email", ""),