api: check phone uniqueness at user serializer validation (#83700)

src/authentic2/api_views.py

@@ -68,7 +68,7 @@ from .journal_event_types import (
     UserNotificationInactivity,
     UserRegistration,
 )
-from .models import APIClient, Attribute, PasswordReset, Service
+from .models import APIClient, Attribute, AttributeValue, PasswordReset, Service
 from .passwords import get_password_checker, get_password_strength
 from .utils import hooks
 from .utils import misc as utils_misc
@@ -434,6 +434,26 @@ class BaseUserSerializer(serializers.ModelSerializer):
                     hasher.safe_summary(attrs.get('hashed_password'))
                 except Exception:
                     errors['hashed_password'] = 'hash format error'
+        authenticator = utils_misc.get_password_authenticator()
+        if authenticator.is_phone_authn_active and (
+            value := attrs.get('attributes', {}).get(authenticator.phone_identifier_field.name, None)
+        ):
+            qs = AttributeValue.objects.filter(
+                attribute=authenticator.phone_identifier_field,
+                content=value,
+            )
+            if self.instance:
+                qs.exclude(object_id=self.instance.id)
+            # manage ou- or global-uniqueness settings
+            ou = attrs.get('ou', None) or get_default_ou()
+            if not app_settings.A2_PHONE_IS_UNIQUE:
+                if not ou.phone_is_unique:
+                    qs = qs.none()
+                else:
+                    qs = qs.filter(object_id__in=User.objects.filter(ou=ou))
+            if qs.exists():
+                errors['attributes'] = _('This phone number identifier is already used.')
+
         if errors:
             raise serializers.ValidationError(errors)
         return attrs

tests/api/test_all.py

@@ -1441,6 +1441,82 @@ def test_api_authn_healthcheck(app, settings, superuser, simple_user):
     app.get('/api/authn-healthcheck/', status=403)
 
 
+def test_api_users_create_phone_identifier_unique(settings, app, admin, phone_activated_authn, simple_user):
+    simple_user.attributes.phone = '+33122334455'
+    simple_user.save()
+    settings.A2_PHONE_IS_UNIQUE = True
+    payload = {
+        'username': 'janedoe',
+        'email': 'jane.doe@nowhere.null',
+        'first_name': 'Jane',
+        'last_name': 'Doe',
+        'email_verified': True,
+        'phone': '+33122334455',
+    }
+    headers = basic_authorization_header(admin)
+    resp = app.post_json('/api/users/', headers=headers, params=payload, status=400)
+    assert resp.json['errors']['attributes'] == ['This phone number identifier is already used.']
+
+
+def test_api_users_create_phone_identifier_unique_by_ou(
+    settings, app, admin, phone_activated_authn, simple_user, ou1, ou2
+):
+    ou1.phone_is_unique = ou2.phone_is_unique = True
+    ou1.save()
+    ou2.save()
+    simple_user.attributes.phone = '+33122334455'
+    simple_user.ou = ou1
+    simple_user.save()
+    usercount = User.objects.count()
+    settings.A2_PHONE_IS_UNIQUE = False
+    payload = {
+        'username': 'janedoe',
+        'email': 'jane.doe@nowhere.null',
+        'first_name': 'Jane',
+        'last_name': 'Doe',
+        'ou': 'ou1',
+        'email_verified': True,
+        'phone': '+33122334455',
+    }
+    headers = basic_authorization_header(admin)
+    resp = app.post_json('/api/users/', headers=headers, params=payload, status=400)
+    assert resp.json['errors']['attributes'] == ['This phone number identifier is already used.']
+    assert set(
+        AttributeValue.objects.filter(
+            attribute=phone_activated_authn.phone_identifier_field,
+            content='+33122334455',
+        ).values_list('object_id', flat=True)
+    ) == {simple_user.id}
+    assert User.objects.count() == usercount
+
+    # change ou, where phone number isn't taken yet
+    payload['ou'] = 'ou2'
+    resp = app.post_json('/api/users/', headers=headers, params=payload, status=201)
+    new_id = resp.json['id']
+    assert new_id != simple_user.id
+    assert set(
+        AttributeValue.objects.filter(
+            attribute=phone_activated_authn.phone_identifier_field,
+            content='+33122334455',
+        ).values_list('object_id', flat=True)
+    ) == {simple_user.id, new_id}
+    assert User.objects.count() == usercount + 1
+
+    # trying to create yet another user in that same last with the same phone number should fail:
+    payload['username'] = 'bobdoe'
+    payload['email'] = 'bobdoe@nowhere.null'
+    resp = app.post_json('/api/users/', headers=headers, params=payload, status=400)
+    assert resp.json['errors']['attributes'] == ['This phone number identifier is already used.']
+    # no new phone attribute created
+    assert set(
+        AttributeValue.objects.filter(
+            attribute=phone_activated_authn.phone_identifier_field,
+            content='+33122334455',
+        ).values_list('object_id', flat=True)
+    ) == {simple_user.id, new_id}
+    assert User.objects.count() == usercount + 1
+
+
 def test_api_users_create_no_phone_model_field_writes(settings, app, admin, phone_activated_authn):
     payload = {
         'username': 'janedoe',