Limit OTP Attempts

drfpasswordless/__version__.py

@@ -1,3 +1,3 @@
-VERSION = (1, 6, 3)
+VERSION = (1, 6, 4)
 
 __version__ = '.'.join(map(str, VERSION))

drfpasswordless/migrations/0006_callbacktoken_attempts.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.25 on 2025-03-20 05:46
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('drfpasswordless', '0005_auto_20201117_0410'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='callbacktoken',
+            name='attempts',
+            field=models.IntegerField(default=0),
+        ),
+    ]

drfpasswordless/models.py

@@ -66,6 +66,16 @@ class CallbackToken(AbstractBaseCallbackToken):
 
     key = models.CharField(default=generate_numeric_token, max_length=6)
     type = models.CharField(max_length=20, choices=TOKEN_TYPES)
+    attempts = models.IntegerField(default=0)
 
     class Meta(AbstractBaseCallbackToken.Meta):
         verbose_name = 'Callback Token'
+
+    def increment_attempts(self):
+        """
+        Increment the number of attempts and deactivate if max attempts reached
+        """
+        self.attempts += 1
+        if self.attempts > 3:  # Max 3 attempts
+            self.is_active = False
+        self.save()

drfpasswordless/serializers.py

@@ -178,7 +178,6 @@ class AbstractBaseCallbackTokenSerializer(serializers.Serializer):
     token = TokenField(
         min_length=api_settings.PASSWORDLESS_TOKEN_LENGTH,
         max_length=api_settings.PASSWORDLESS_TOKEN_LENGTH,
-        validators=[token_age_validator]
     )
 
     def validate_alias(self, attrs):
@@ -209,6 +208,18 @@ def validate(self, attrs):
             alias_type, alias = self.validate_alias(attrs)
             callback_token = attrs.get('token', None)
             user = User.objects.get(**{alias_type+'__iexact': alias})
+
+
+            #increment the attempt
+            token = CallbackToken.objects.filter(**{'user': user,
+                                                 'type': CallbackToken.TOKEN_TYPE_AUTH,
+                                                 'is_active': True}).first()
+
+            if token:
+                token.increment_attempts()
+
+            validate_token_age(callback_token)
+
             token = CallbackToken.objects.get(**{'user': user,
                                                  'key': callback_token,
                                                  'type': CallbackToken.TOKEN_TYPE_AUTH,
@@ -238,7 +249,7 @@ def validate(self, attrs):
                 msg = _('Invalid Token')
                 raise serializers.ValidationError(msg)
         except CallbackToken.DoesNotExist:
-            msg = _('Invalid alias parameters provided.')
+            msg = _('Invalid OTP or OTP may be expired')
             raise serializers.ValidationError(msg)
         except User.DoesNotExist:
             msg = _('Invalid user alias parameters provided.')