Add OTP brute-force protection and single-use OTP

auth_token/authorization_request/backends.py

@@ -1,5 +1,9 @@
 from auth_token.config import settings
-from auth_token.utils import create_otp, deactivate_otp, get_valid_otp, otp_key_generator_factory
+from auth_token.exceptions import MaxOTPAttemptsExceeded
+from auth_token.utils import (
+    cancel_authorization_request, create_otp, deactivate_otp,
+    get_valid_otp, otp_key_generator_factory,
+)
 from auth_token.models import MobileDevice
 
 
@@ -74,15 +78,29 @@ def initialize(self, authorization_request):
         otp = self._create_otp(authorization_request)
         authorization_request.secret_key = otp.secret_key
 
-    def authenticate(self, authorization_request, otp_secret_key=None, **kwargs):
+    def authenticate(self, authorization_request, otp_secret_key=None, deactivate_otp_on_success=True, **kwargs):
         if otp_secret_key is None:
             return False
 
         if (settings.AUTHORIZATION_REQUEST_OTP_DEBUG_CODE
                 and settings.AUTHORIZATION_REQUEST_OTP_DEBUG_CODE == otp_secret_key):
-            otp_secret_key = None
+            return True
+
+        otp = get_valid_otp(
+            authorization_request.slug,
+            key=otp_secret_key,
+            related_objects=[authorization_request],
+            deactivate=deactivate_otp_on_success,
+        )
+
+        if otp is None and settings.MAX_OTP_ATTEMPTS is not None:
+            authorization_request.change_and_save(
+                failed_attempts=authorization_request.failed_attempts + 1
+            )
+            if authorization_request.failed_attempts >= settings.MAX_OTP_ATTEMPTS:
+                cancel_authorization_request(authorization_request)
+                raise MaxOTPAttemptsExceeded
 
-        otp = get_valid_otp(authorization_request.slug, key=otp_secret_key, related_objects=[authorization_request])
         return otp is not None
 
     def _deactivate(self, authorization_request):

auth_token/config.py

@@ -49,6 +49,7 @@
     'OTP_DEFAULT_KEY_GENERATOR': 'auth_token.utils.generate_otp_key',  # Default key generator for OTP
     'OTP_EXPIRATION_RETENTION_PERIOD': 0,  # Expired tokens are deleted immediately
 
+    'MAX_OTP_ATTEMPTS': None,  # Maximum number of failed OTP attempts before cancelling authorization request
     'DEFAULT_AUTHORIZATION_REQUEST_AGE': 60 * 60,  # Default authorization request expiration time (default: 1 hour),
     'AUTHORIZATION_REQUEST_PRESERVE_AGE': 60 * 60 * 24 * 7,  # Keep expired authorization requests in database
                                                              # (default: 7 days)

auth_token/contrib/is_core_auth/views.py

@@ -1,10 +1,12 @@
 from urllib.parse import quote_plus
 
 from django import forms
+from django.contrib import messages
 from django.contrib.auth import login as auth_login
 from django.core.exceptions import ValidationError
 from django.forms.utils import ErrorList
 from django.http import HttpResponseRedirect
+from django.utils.translation import gettext as _
 from django.views.generic.base import RedirectView
 from django.urls import reverse
 
@@ -17,6 +19,7 @@
 from auth_token.contrib.common.views import LogoutView as _LogoutView
 from auth_token.contrib.common.views import LoginCodeVerificationView as _LoginCodeVerificationView
 from auth_token.contrib.is_core_auth.forms import LoginCodeVerificationForm
+from auth_token.exceptions import MaxOTPAttemptsExceeded
 from auth_token.utils import create_authorization_request, grant_authorization_request, login, takeover
 
 from is_core.generic_views.base import DefaultCoreViewMixin
@@ -94,6 +97,17 @@ class LoginCodeVerificationView(_LoginCodeVerificationView):
     template_name = 'is_core/login.html'
     form_class = LoginCodeVerificationForm
 
+    def post(self, request, *args, **kwargs):
+        try:
+            return super().post(request, *args, **kwargs)
+        except MaxOTPAttemptsExceeded:
+            messages.error(request, _('Maximum number of verification attempts reached. Please log in again.'))
+            return HttpResponseRedirect('{url}?{redirect_field_name}={value}'.format(
+                url=reverse('IS:login'),
+                redirect_field_name=self.redirect_field_name,
+                value=self.get_redirect_url(),
+            ))
+
     def form_valid(self, form):
         self.log_successful_request()
         grant_authorization_request(form.get_authorization_request())

auth_token/exceptions.py

@@ -1,2 +1,6 @@
 class KeyGeneratorError(Exception):
     pass
+
+
+class MaxOTPAttemptsExceeded(Exception):
+    pass

auth_token/locale/cs/LC_MESSAGES/django.po

@@ -63,6 +63,10 @@ msgstr "kód"
 msgid "The inserted value does not correspond to the sent code."
 msgstr "Vložená hodnota neodpovídá zaslanému kódu."
 
+#: contrib/is_core_auth/views.py:104
+msgid "Maximum number of verification attempts reached. Please log in again."
+msgstr "Dosáhli jste maximálního počtu pokusů o ověření. Přihlaste se prosím znovu."
+
 #: contrib/ms_sso/views.py:35 contrib/ms_sso/views.py:40
 msgid "Microsoft SSO login was unsuccessful, please try it again"
 msgstr "Přihlášení pomocí Microsoft účtu bylo neúspěšné, prosím zkuste to znovu"

auth_token/migrations/0017_migration.py

@@ -0,0 +1,16 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('auth_token', '0016_migration'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='authorizationrequest',
+            name='failed_attempts',
+            field=models.PositiveIntegerField(default=0, verbose_name='failed attempts'),
+        ),
+    ]

auth_token/models.py

@@ -465,6 +465,10 @@ class AuthorizationRequest(SmartModel):
         null=True,
         blank=True
     )
+    failed_attempts = models.PositiveIntegerField(
+        verbose_name=_('failed attempts'),
+        default=0,
+    )
     related_objects = GenericManyToManyField()
 
     class Meta:

auth_token/version.py

@@ -1,4 +1,4 @@
-VERSION = (0, 6, 5)
+VERSION = (0, 7, 0)
 
 
 def get_version():

example/dj/apps/app/tests/is_core.py

@@ -283,3 +283,33 @@ def test_user_should_not_be_logged_in_two_factor_for_invalid_code(self, user):
             code_check_resp._container[0].decode('utf8')
         )
         assert_false(code_check_resp.wsgi_request.token.is_authenticated)
+
+    @override_settings(AUTH_TOKEN_TWO_FACTOR_ENABLED=True)
+    @override_settings(
+        AUTH_TOKEN_TWO_FACTOR_SENDING_FUNCTION='app.tests.is_core.UILoginISCoreTestCase.send_two_factor_token')
+    @override_settings(AUTH_TOKEN_MAX_OTP_ATTEMPTS=3)
+    @data_consumer('create_user')
+    def test_user_should_be_redirected_to_login_after_max_otp_attempts_exceeded(self, user):
+        login_resp = self.post(self.UI_TWO_FACTOR_LOGIN_URL, {'username': 'test', 'password': 'test'})
+
+        assert_http_redirect(login_resp)
+        assert_false(login_resp.wsgi_request.user.is_authenticated)
+
+        # First two wrong attempts return form error without exceeding the limit
+        for __ in range(2):
+            resp = self.post(self.UI_CODE_CHECK_LOGIN_URL, {'code': 'wrong_code'})
+            assert_http_ok(resp)
+            assert_false(resp.wsgi_request.token.is_authenticated)
+
+        # Third wrong attempt reaches MAX_OTP_ATTEMPTS and triggers a redirect to login
+        code_check_resp = self.post(self.UI_CODE_CHECK_LOGIN_URL, {'code': 'wrong_code'})
+
+        assert_http_redirect(code_check_resp)
+        assert_in('/is_core/login/', code_check_resp['Location'])
+        assert_false(code_check_resp.wsgi_request.token.is_authenticated)
+
+        login_page_resp = self.get(code_check_resp['Location'])
+        assert_in(
+            _('Maximum number of verification attempts reached. Please log in again.'),
+            login_page_resp._container[0].decode('utf8')
+        )

example/dj/apps/app/tests/utils.py

@@ -8,6 +8,8 @@
 from django.core.exceptions import PermissionDenied
 from freezegun import freeze_time
 
+from auth_token.exceptions import MaxOTPAttemptsExceeded
+
 from django.core.management import call_command
 from django.utils.timezone import now
 from django.test import override_settings
@@ -574,6 +576,46 @@ def test_check_authorization_request_with_otp_should_return_right_value(self, us
                                                      otp_secret_key=authorization_request.secret_key))
             assert_equal(authorization_request.state, AuthorizationRequestState.EXPIRED)
 
+    @freeze_time(now())
+    @data_consumer('create_user')
+    def test_check_authorization_request_with_otp_should_deactivate_otp_on_success(self, user):
+        authorization_request = create_authorization_request(
+            'test', user, 'test',
+            backend_path='auth_token.authorization_request.backends.OTPAuthorizationRequestBackend'
+        )
+        secret_key = authorization_request.secret_key
+        # OTP is consumed on successful check
+        assert_true(check_authorization_request(authorization_request, otp_secret_key=secret_key))
+        # Same OTP can no longer be used
+        assert_false(check_authorization_request(authorization_request, otp_secret_key=secret_key))
+
+    @freeze_time(now())
+    @data_consumer('create_user')
+    @override_settings(AUTH_TOKEN_MAX_OTP_ATTEMPTS=3)
+    def test_check_authorization_request_with_otp_should_cancel_after_max_failed_attempts(self, user):
+        authorization_request = create_authorization_request(
+            'test', user, 'test',
+            backend_path='auth_token.authorization_request.backends.OTPAuthorizationRequestBackend'
+        )
+        # First 2 failed attempts are within the limit
+        assert_false(check_authorization_request(authorization_request, otp_secret_key='wrong1'))
+        assert_false(check_authorization_request(authorization_request, otp_secret_key='wrong2'))
+        # 3rd failed attempt reaches MAX_OTP_ATTEMPTS=3, request is cancelled
+        with assert_raises(MaxOTPAttemptsExceeded):
+            check_authorization_request(authorization_request, otp_secret_key='wrong3')
+        assert_equal(authorization_request.refresh_from_db().state, AuthorizationRequestState.CANCELLED)
+
+    @freeze_time(now())
+    @data_consumer('create_user')
+    def test_check_authorization_request_with_otp_should_not_cancel_when_max_attempts_not_set(self, user):
+        authorization_request = create_authorization_request(
+            'test', user, 'test',
+            backend_path='auth_token.authorization_request.backends.OTPAuthorizationRequestBackend'
+        )
+        for _ in range(10):
+            assert_false(check_authorization_request(authorization_request, otp_secret_key='wrong'))
+        assert_equal(authorization_request.refresh_from_db().state, AuthorizationRequestState.WAITING)
+
     @freeze_time(now())
     @data_consumer('create_user')
     def test_authorization_reset_create_new_otp_key_and_increase_expiration(self, user):