Only validate token against chosen device (jazzband#473)

tests/test_views_login.py

@@ -254,8 +254,8 @@ def test_throttle_with_generator(self, mock_signal):
         response = self._post({'token-otp_token': totp(device.bin_key),
                                'login_view-current_step': 'token'})
         self.assertEqual(response.context_data['wizard']['form'].errors,
-                         {'__all__': ['Invalid token. Please make sure you '
-                                      'have entered it correctly.']})
+                         {'__all__': ['Verification temporarily disabled because '
+                                      'of 1 failed attempt, please try again soon.']})
 
     @mock.patch('two_factor.gateways.fake.Fake')
     @mock.patch('two_factor.views.core.signals.user_verified.send')
@@ -312,8 +312,9 @@ def test_with_backup_phone(self, mock_signal, fake):
                                for i in [-1, 0]])
 
             # Valid token should be accepted.
-            response = self._post({'token-otp_token': totp(device.bin_key),
-                                   'login_view-current_step': 'token'})
+            response = self._post({'token-otp_token': totp_str(device.bin_key),
+                                   'login_view-current_step': 'token',
+                                   'device_id': device.persistent_id})
             self.assertRedirects(response, resolve_url(settings.LOGIN_REDIRECT_URL))
             self.assertEqual(device.persistent_id,
                              self.client.session.get(DEVICE_ID_SESSION_KEY))
@@ -355,6 +356,27 @@ def test_with_backup_token(self, mock_signal):
         # Check that the signal was fired.
         mock_signal.assert_called_with(sender=mock.ANY, request=mock.ANY, user=user, device=device)
 
+    def test_totp_token_does_not_impact_backup_token(self):
+        user = self.create_user()
+        backup_device = user.staticdevice_set.create(name='backup')
+        backup_device.token_set.create(token='abcdef123')
+        totp_device = user.totpdevice_set.create(name='default', key=random_hex())
+
+        response = self._post({'auth-username': 'bouke@example.com',
+                               'auth-password': 'secret',
+                               'login_view-current_step': 'auth'})
+        self.assertContains(response, 'Token:')
+
+        backup_device.refresh_from_db()
+        self.assertEqual(backup_device.throttling_failure_count, 0)
+        response = self._post({'token-otp_token': totp_str(totp_device.bin_key),
+                               'login_view-current_step': 'token'})
+        self.assertRedirects(response, resolve_url(settings.LOGIN_REDIRECT_URL))
+        self.assertEqual(self.client.session['_auth_user_id'], str(user.pk))
+
+        backup_device.refresh_from_db()
+        self.assertEqual(backup_device.throttling_failure_count, 0)
+
     @mock.patch('two_factor.views.utils.logger')
     def test_reset_wizard_state(self, mock_logger):
         self.create_user()

two_factor/forms.py

@@ -3,7 +3,9 @@
 
 from django import forms
 from django.conf import settings
+from django.core.exceptions import ObjectDoesNotExist
 from django.utils.translation import gettext_lazy as _
+from django_otp import devices_for_user
 from django_otp.forms import OTPAuthenticationFormMixin
 from django_otp.oath import totp
 from django_otp.plugins.otp_totp.models import TOTPDevice
@@ -104,12 +106,16 @@ class DisableForm(forms.Form):
 
 
 class AuthenticationTokenForm(OTPAuthenticationFormMixin, forms.Form):
-    otp_token = forms.IntegerField(label=_("Token"), min_value=1,
-                                   max_value=int('9' * totp_digits()))
-
-    otp_token.widget.attrs.update({'autofocus': 'autofocus',
-                                   'inputmode': 'numeric',
-                                   'autocomplete': 'one-time-code'})
+    otp_token = forms.RegexField(label=_("Token"),
+                                 regex=r'^[0-9]*$',
+                                 min_length=totp_digits(),
+                                 max_length=totp_digits())
+    otp_token.widget.attrs.update({
+        'autofocus': 'autofocus',
+        'pattern': '[0-9]*',  # hint to show numeric keyboard for on-screen keyboards
+        'autocomplete': 'one-time-code',
+    })
+    device_id = forms.CharField(widget=forms.HiddenInput(), required=False)
 
     # Our authentication form has an additional submit button to go to the
     # backup token form. When the `required` attribute is set on an input
@@ -119,15 +125,21 @@ class AuthenticationTokenForm(OTPAuthenticationFormMixin, forms.Form):
     # its own `<form>`.
     use_required_attribute = False
 
+    error_messages = {
+        'invalid_device_id': _('Device id is not valid.'),
+    }
+
     def __init__(self, user, initial_device, **kwargs):
         """
-        `initial_device` is either the user's default device, or the backup
-        device when the user chooses to enter a backup token. The token will
-        be verified against all devices, it is not limited to the given
-        device.
+        `initial_device` is either the user's default device a backup device
+        when the user chooses to enter a backup token.
         """
         super().__init__(**kwargs)
         self.user = user
+        self.initial_device = initial_device
+        if initial_device:
+            self.fields['device_id'].initial = initial_device.persistent_id
+        self.device_cache = None
 
         # Add a field to remeber this browser.
         if getattr(settings, 'TWO_FACTOR_REMEMBER_COOKIE_AGE', None):
@@ -147,6 +159,22 @@ def __init__(self, user, initial_device, **kwargs):
                 label=label
             )
 
+    def clean_device_id(self):
+        if self.data.get("device_id"):
+            try:
+                for user_device in devices_for_user(self.user):
+                    if user_device.persistent_id == self.data["device_id"]:
+                        self.device_cache = user_device
+                        break
+            except ObjectDoesNotExist:
+                raise forms.ValidationError(self.error_messages['invalid_device_id'])
+
+    def _chosen_device(self, user):
+        if self.device_cache:
+            return self.device_cache
+        else:
+            return self.initial_device
+
     def clean(self):
         self.clean_otp(self.user)
         return self.cleaned_data