Added throttling to token generation (Closes #48)

src/django_otp/models.py

@@ -159,6 +159,19 @@ def is_interactive(self):
         """
         return not hasattr(self.generate_challenge, 'stub')
 
+    def generate_is_allowed(self):
+        """
+        Checks whether it is permissible to call :meth:`generate_challenge`. If
+        it is allowed, returns ``(True, None)``. Otherwise returns ``(False,
+        data_dict)``, where ``data_dict`` contains extra information, defined
+        by the implementation.
+
+        This method can be used to implement throttling of token generation for
+        interactive devices. Client code should check this method before calling
+        :meth:`generate_challenge` and report problems to the user.
+        """
+        return (True, None)
+
     def generate_challenge(self):
         """
         Generates a challenge value that the user will need to produce a token.
@@ -293,6 +306,68 @@ class VerifyNotAllowed:
     N_FAILED_ATTEMPTS = 'N_FAILED_ATTEMPTS'
 
 
+class GenerationCooldownMixin(models.Model):
+    """
+    Mixin class for models requiring a cooldown duration between generating a challenge.
+    """
+
+    last_generated_timestamp = models.DateTimeField(
+        null=True,
+        blank=True,
+        help_text="The last time a token was generated for this device.",
+    )
+
+    def generate_is_allowed(self):
+        """
+        Checks if the time since the last token generation is greater than
+        configured (in seconds).
+        """
+        dt_now = timezone.now()
+        if (
+            not self.last_generated_timestamp
+            or (dt_now - self.last_generated_timestamp).total_seconds()
+            > self.get_cooldown_duration()
+        ):
+            return super().generate_is_allowed()
+        else:
+            return False, {
+                'reason': "COOLDOWN_DURATION_PENDING",
+                'next_generation_at': dt_now
+                + timedelta(seconds=self.get_cooldown_duration()),
+            }
+
+    def cooldown_reset(self, commit=True):
+        """
+        Call this method to reset cooldown (normally when a successful
+        verification).
+
+        Pass 'commit=False' to avoid calling self.save().
+        """
+        self.last_generated_timestamp = None
+
+        if commit:
+            self.save()
+
+    def verify_token(self, token):
+        """
+        Reset the throttle if the token is valid.
+        """
+        verified = super().verify_token(token)
+        if verified:
+            self.cooldown_reset()
+        return verified
+
+    @cached_property
+    def cooldown_enabled(self):
+        return self.get_cooldown_duration() > 0
+
+    def get_cooldown_duration(self):
+        raise NotImplementedError()
+
+    class Meta:
+        abstract = True
+
+
 class ThrottlingMixin(models.Model):
     """
     Mixin class for models that want throttling behaviour.
@@ -311,7 +386,10 @@ class ThrottlingMixin(models.Model):
         null=True,
         blank=True,
         default=None,
-        help_text="A timestamp of the last failed verification attempt. Null if last attempt succeeded.",
+        help_text=(
+            "A timestamp of the last failed verification attempt. Null if last attempt"
+            " succeeded."
+        ),
     )
 
     throttling_failure_count = models.PositiveIntegerField(

src/django_otp/plugins/otp_email/migrations/0005_emaildevice_last_generated_timestamp.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.16 on 2023-05-12 11:37
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('otp_email', '0004_throttling'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='emaildevice',
+            name='last_generated_timestamp',
+            field=models.DateTimeField(blank=True, help_text='The last time a token was generated for this device.', null=True),
+        ),
+    ]

src/django_otp/plugins/otp_email/models.py

@@ -1,9 +1,15 @@
+from django.contrib.humanize.templatetags.humanize import naturaltime
 from django.core.mail import send_mail
 from django.db import models
 from django.template import Context, Template
 from django.template.loader import get_template
+from django.utils import timezone
 
-from django_otp.models import SideChannelDevice, ThrottlingMixin
+from django_otp.models import (
+    GenerationCooldownMixin,
+    SideChannelDevice,
+    ThrottlingMixin,
+)
 from django_otp.util import hex_validator, random_hex
 
 from .conf import settings
@@ -19,7 +25,7 @@ def key_validator(value):  # pragma: no cover
     return hex_validator()(value)
 
 
-class EmailDevice(ThrottlingMixin, SideChannelDevice):
+class EmailDevice(GenerationCooldownMixin, ThrottlingMixin, SideChannelDevice):
     """
     A :class:`~django_otp.models.SideChannelDevice` that delivers a token to
     the email address saved in this object or alternatively to the user's
@@ -57,34 +63,50 @@ def generate_challenge(self, extra_context=None):
         :type extra_context: dict
 
         """
-        self.generate_token(valid_secs=settings.OTP_EMAIL_TOKEN_VALIDITY)
-
-        context = {'token': self.token, **(extra_context or {})}
-        if settings.OTP_EMAIL_BODY_TEMPLATE:
-            body = Template(settings.OTP_EMAIL_BODY_TEMPLATE).render(Context(context))
-        else:
-            body = get_template(settings.OTP_EMAIL_BODY_TEMPLATE_PATH).render(context)
-
-        if settings.OTP_EMAIL_BODY_HTML_TEMPLATE:
-            body_html = Template(settings.OTP_EMAIL_BODY_HTML_TEMPLATE).render(
-                Context(context)
-            )
-        elif settings.OTP_EMAIL_BODY_HTML_TEMPLATE_PATH:
-            body_html = get_template(settings.OTP_EMAIL_BODY_HTML_TEMPLATE_PATH).render(
-                context
+        generate_allowed, data_dict = self.generate_is_allowed()
+        if generate_allowed:
+            self.generate_token(valid_secs=settings.OTP_EMAIL_TOKEN_VALIDITY)
+            self.last_generated_timestamp = timezone.now()
+
+            context = {'token': self.token, **(extra_context or {})}
+            if settings.OTP_EMAIL_BODY_TEMPLATE:
+                body = Template(settings.OTP_EMAIL_BODY_TEMPLATE).render(
+                    Context(context)
+                )
+            else:
+                body = get_template(settings.OTP_EMAIL_BODY_TEMPLATE_PATH).render(context)
+
+            if settings.OTP_EMAIL_BODY_HTML_TEMPLATE:
+                body_html = Template(settings.OTP_EMAIL_BODY_HTML_TEMPLATE).render(
+                    Context(context)
+                )
+            elif settings.OTP_EMAIL_BODY_HTML_TEMPLATE_PATH:
+                body_html = get_template(settings.OTP_EMAIL_BODY_HTML_TEMPLATE_PATH).render(
+                    context
+                )
+            else:
+                body_html = None
+
+            send_mail(
+                str(settings.OTP_EMAIL_SUBJECT),
+                body,
+                settings.OTP_EMAIL_SENDER,
+                [self.email or self.user.email],
+                html_message=body_html,
             )
-        else:
-            body_html = None
-
-        send_mail(
-            str(settings.OTP_EMAIL_SUBJECT),
-            body,
-            settings.OTP_EMAIL_SENDER,
-            [self.email or self.user.email],
-            html_message=body_html,
-        )
 
-        message = "sent by email"
+            message = "sent by email"
+        else:
+            if data_dict['reason'] == 'COOLDOWN_DURATION_PENDING':
+                next_generation_naturaltime = naturaltime(
+                    data_dict['next_generation_at']
+                )
+                message = (
+                    "Token generation cooldown period has not expired yet. Next"
+                    f" generation allowed {next_generation_naturaltime}"
+                )
+            else:
+                message = "Token generation is not allowed at this time"
 
         return message
 
@@ -101,3 +123,6 @@ def verify_token(self, token):
             verified = False
 
         return verified
+
+    def get_cooldown_duration(self):
+        return settings.OTP_EMAIL_GENERATION_INTERVAL

src/django_otp/plugins/otp_email/tests.py

@@ -7,7 +7,11 @@
 from django.test.utils import override_settings
 
 from django_otp.forms import OTPAuthenticationForm
-from django_otp.tests import TestCase, ThrottlingTestMixin
+from django_otp.tests import (
+    GenerationThrottlingTestMixin,
+    TestCase,
+    ThrottlingTestMixin,
+)
 
 from .models import EmailDevice
 
@@ -235,3 +239,12 @@ def valid_token(self):
 
     def invalid_token(self):
         return -1
+
+
+@override_settings(
+    OTP_EMAIL_GENERATION_INTERVAL=1,
+)
+class GenerationThrottlingTestCase(
+    EmailDeviceMixin, GenerationThrottlingTestMixin, TestCase
+):
+    pass

src/django_otp/tests.py

@@ -9,6 +9,7 @@
 
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import AnonymousUser
+from django.core import mail
 from django.core.management import call_command
 from django.core.management.base import CommandError
 from django.db import IntegrityError, connection
@@ -164,6 +165,45 @@ def test_verify_is_allowed(self):
             self.assertEqual(data3, None)
 
 
+class GenerationThrottlingTestMixin:
+    def setUp(self):
+        self.device = None
+
+    def test_cooldown_imposed_after_successful_generation(self):
+        message = self.device.generate_challenge()
+        self.assertEqual(message, 'sent by email')
+        message = self.device.generate_challenge()
+        self.assertTrue(
+            message.startswith('Token generation cooldown period has not expired yet.')
+        )
+
+    def test_allow_generation_after_cooldown(self):
+        self.assertEqual(len(mail.outbox), 0)
+        # First generation is allowed
+        self.device.generate_challenge()
+        # Assert that an email is sent
+        self.assertEqual(len(mail.outbox), 1)
+
+        with freeze_time():
+            # Second generation, within cooldown period
+            message = self.device.generate_challenge()
+            self.assertTrue(
+                message.startswith(
+                    'Token generation cooldown period has not expired yet.'
+                )
+            )
+            # Assert that no email is sent
+            self.assertEqual(len(mail.outbox), 1)
+
+        with freeze_time() as frozen_time:
+            # Third generation after cooldown period
+            frozen_time.tick(delta=timedelta(seconds=1.1))
+            message = self.device.generate_challenge()
+            self.assertEqual(message, 'sent by email')
+            # Assert that a second email is sent
+            self.assertEqual(len(mail.outbox), 2)
+
+
 @override_settings(OTP_STATIC_THROTTLE_FACTOR=0)
 class APITestCase(TestCase):
     def setUp(self):