Merge pull request from GHSA-vhr6-pvjm-9qwf

Bouke · web-flow · commit 454fd9842fa6 · 2020-07-08T21:11:38.000+02:00
Discontinue storing credentials in the session store
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,4 +1,12 @@
 ## [Unreleased]
+### Added
+- It is possible to set a timeout between a user authenticiating in the LoginView and them needing to re-authenticate. By default this is 10 minutes.
+
+### Removed
+- The final step in the LoginView no longer re-validates a user's credentials
+
+### Changed
+- Security Fix: LoginView no longer stores credentials in plaintext in the session store
 
 ## 1.11.0 - 2020-03-13
 ### Added
diff --git a/docs/configuration.rst b/docs/configuration.rst
@@ -73,6 +73,13 @@ General Settings
      `the upstream ticket`_). Don't set this option to 8 unless all of your
      users use a 8 digit compatible token generator app.
 
+``TWO_FACTOR_LOGIN_TIMEOUT`` (default ``600``)
+  The number of seconds between a user successfully passing the "authentication"
+  step (usually by entering a valid username and password) and them having to
+  restart the login flow and re-authenticate. This ensures that users can't sit
+  indefinately in a state of having entered their password successfully but not
+  having passed two factor authentication. Set to ``0`` to disable.
+
 ``PHONENUMBER_DEFAULT_REGION`` (default: ``None``)
   The default region for parsing phone numbers. If your application's primary
   audience is a certain country, setting the region to that country allows
diff --git a/tests/test_views_login.py b/tests/test_views_login.py
@@ -1,3 +1,4 @@
+import json
 from unittest import mock
 
 from django.conf import settings
@@ -88,6 +89,100 @@ def test_valid_login_with_disallowed_external_redirect(self):
              'login_view-current_step': 'auth'})
         self.assertRedirects(response, reverse('two_factor:profile'), fetch_redirect_response=False)
 
+    @mock.patch('two_factor.views.core.time')
+    def test_valid_login_primary_key_stored(self, mock_time):
+        mock_time.time.return_value = 12345.12
+        user = self.create_user()
+        user.totpdevice_set.create(name='default',
+                                   key=random_hex_str())
+
+        response = self._post({'auth-username': 'bouke@example.com',
+                               'auth-password': 'secret',
+                               'login_view-current_step': 'auth'})
+        self.assertContains(response, 'Token:')
+
+        self.assertEqual(self.client.session['wizard_login_view']['user_pk'], str(user.pk))
+        self.assertEqual(
+            self.client.session['wizard_login_view']['user_backend'],
+            'django.contrib.auth.backends.ModelBackend')
+        self.assertEqual(self.client.session['wizard_login_view']['authentication_time'], 12345)
+
+    @mock.patch('two_factor.views.core.time')
+    def test_valid_login_post_auth_session_clear_of_form_data(self, mock_time):
+        mock_time.time.return_value = 12345.12
+        user = self.create_user()
+        user.totpdevice_set.create(name='default',
+                                   key=random_hex_str())
+
+        response = self._post({'auth-username': 'bouke@example.com',
+                               'auth-password': 'secret',
+                               'login_view-current_step': 'auth'})
+        self.assertContains(response, 'Token:')
+
+        self.assertEqual(self.client.session['wizard_login_view']['user_pk'], str(user.pk))
+        self.assertEqual(self.client.session['wizard_login_view']['step'], 'token')
+        self.assertEqual(self.client.session['wizard_login_view']['step_data'], {'auth': None})
+        self.assertEqual(self.client.session['wizard_login_view']['step_files'], {'auth': {}})
+        self.assertEqual(self.client.session['wizard_login_view']['validated_step_data'], {})
+
+    @mock.patch('two_factor.views.core.logger')
+    @mock.patch('two_factor.views.core.time')
+    def test_valid_login_expired(self, mock_time, mock_logger):
+        mock_time.time.return_value = 12345.12
+        user = self.create_user()
+        device = user.totpdevice_set.create(name='default',
+                                            key=random_hex_str())
+
+        response = self._post({'auth-username': 'bouke@example.com',
+                               'auth-password': 'secret',
+                               'login_view-current_step': 'auth'})
+        self.assertContains(response, 'Token:')
+
+        self.assertEqual(self.client.session['wizard_login_view']['user_pk'], str(user.pk))
+        self.assertEqual(
+            self.client.session['wizard_login_view']['user_backend'],
+            'django.contrib.auth.backends.ModelBackend')
+        self.assertEqual(self.client.session['wizard_login_view']['authentication_time'], 12345)
+
+        mock_time.time.return_value = 20345.12
+
+        response = self._post({'token-otp_token': totp(device.bin_key),
+                               'login_view-current_step': 'token'})
+        self.assertEqual(response.status_code, 200)
+        self.assertNotContains(response, 'Token:')
+        self.assertContains(response, 'Password:')
+        self.assertContains(response, 'Your session has timed out. Please login again.')
+
+        # Check that a message was logged.
+        mock_logger.info.assert_called_with(
+            "User's authentication flow has timed out. The user "
+            "has been redirected to the initial auth form.")
+
+    @override_settings(TWO_FACTOR_LOGIN_TIMEOUT=0)
+    @mock.patch('two_factor.views.core.time')
+    def test_valid_login_no_timeout(self, mock_time):
+        mock_time.time.return_value = 12345.12
+        user = self.create_user()
+        device = user.totpdevice_set.create(name='default',
+                                            key=random_hex_str())
+
+        response = self._post({'auth-username': 'bouke@example.com',
+                               'auth-password': 'secret',
+                               'login_view-current_step': 'auth'})
+        self.assertContains(response, 'Token:')
+
+        self.assertEqual(self.client.session['wizard_login_view']['user_pk'], str(user.pk))
+        self.assertEqual(
+            self.client.session['wizard_login_view']['user_backend'],
+            'django.contrib.auth.backends.ModelBackend')
+        self.assertEqual(self.client.session['wizard_login_view']['authentication_time'], 12345)
+
+        mock_time.time.return_value = 20345.12
+
+        response = self._post({'token-otp_token': 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))
 
     def test_valid_login_with_redirect_authenticated_user(self):
         user = self.create_user()
@@ -251,35 +346,6 @@ 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)
 
-    @mock.patch('two_factor.views.utils.logger')
-    def test_change_password_in_between(self, mock_logger):
-        """
-        When the password of the user is changed while trying to login, should
-        not result in errors. Refs #63.
-        """
-        user = self.create_user()
-        self.enable_otp()
-
-        response = self._post({'auth-username': 'bouke@example.com',
-                               'auth-password': 'secret',
-                               'login_view-current_step': 'auth'})
-        self.assertContains(response, 'Token:')
-
-        # Now, the password is changed. When the form is submitted, the
-        # credentials should be checked again. If that's the case, the
-        # login form should note that the credentials are invalid.
-        user.set_password('secret2')
-        user.save()
-        response = self._post({'login_view-current_step': 'token'})
-        self.assertContains(response, 'Please enter a correct')
-        self.assertContains(response, 'and password.')
-
-        # Check that a message was logged.
-        mock_logger.warning.assert_called_with(
-            "Current step '%s' is no longer valid, returning to last valid "
-            "step in the wizard.",
-            'token')
-
     @mock.patch('two_factor.views.utils.logger')
     def test_reset_wizard_state(self, mock_logger):
         self.create_user()
@@ -332,6 +398,19 @@ def test_missing_management_data(self):
         # view should return HTTP 400 Bad Request
         self.assertEqual(response.status_code, 400)
 
+    def test_no_password_in_session(self):
+        self.create_user()
+        self.enable_otp()
+
+        response = self._post({'auth-username': 'bouke@example.com',
+                               'auth-password': 'secret',
+                               'login_view-current_step': 'auth'})
+        self.assertContains(response, 'Token:')
+
+        session_contents = json.dumps(list(self.client.session.items()))
+
+        self.assertNotIn('secret', session_contents)
+
 
 class BackupTokensTest(UserMixin, TestCase):
     def setUp(self):
diff --git a/two_factor/views/core.py b/two_factor/views/core.py
@@ -2,6 +2,7 @@
 import warnings
 from base64 import b32encode
 from binascii import unhexlify
+import time
 
 import django_otp
 import qrcode
@@ -12,13 +13,15 @@
 from django.contrib.auth.forms import AuthenticationForm
 from django.contrib.auth.views import SuccessURLAllowedHostsMixin
 from django.contrib.sites.shortcuts import get_current_site
-from django.forms import Form
+from django.forms import Form, ValidationError
 from django.http import Http404, HttpResponse, HttpResponseRedirect
 from django.shortcuts import redirect, resolve_url
 from django.urls import reverse
 from django.utils.decorators import method_decorator
+from django.utils.functional import cached_property
 from django.utils.http import is_safe_url
 from django.utils.module_loading import import_string
+from django.utils.translation import gettext as _
 from django.views.decorators.cache import never_cache
 from django.views.decorators.csrf import csrf_protect
 from django.views.decorators.debug import sensitive_post_parameters
@@ -72,6 +75,7 @@ class LoginView(SuccessURLAllowedHostsMixin, IdempotentSessionWizardView):
         'backup': False,
     }
     redirect_authenticated_user = False
+    storage_name = 'two_factor.views.utils.LoginStorage'
 
     def has_token_step(self):
         return default_device(self.get_user())
@@ -80,6 +84,14 @@ def has_backup_step(self):
         return default_device(self.get_user()) and \
             'token' not in self.storage.validated_step_data
 
+    @cached_property
+    def expired(self):
+        login_timeout = getattr(settings, 'TWO_FACTOR_LOGIN_TIMEOUT', 600)
+        if login_timeout == 0:
+            return False
+        expiration_time = self.storage.data.get("authentication_time", 0) + login_timeout
+        return int(time.time()) > expiration_time
+
     condition_dict = {
         'token': has_token_step,
         'backup': has_backup_step,
@@ -90,12 +102,25 @@ def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.user_cache = None
         self.device_cache = None
+        self.show_timeout_error = False
 
     def post(self, *args, **kwargs):
         """
         The user can select a particular device to challenge, being the backup
         devices added to the account.
         """
+        wizard_goto_step = self.request.POST.get('wizard_goto_step', None)
+
+        if wizard_goto_step == 'auth':
+            self.storage.reset()
+
+        if self.expired and self.steps.current != 'auth':
+            logger.info("User's authentication flow has timed out. The user "
+                        "has been redirected to the initial auth form.")
+            self.storage.reset()
+            self.show_timeout_error = True
+            return self.render_goto_step('auth')
+
         # Generating a challenge doesn't require to validate the form.
         if 'challenge_device' in self.request.POST:
             return self.render_goto_step('token')
@@ -152,6 +177,54 @@ def get_form_kwargs(self, step=None):
             }
         return {}
 
+    def get_done_form_list(self):
+        """
+        Return the forms that should be processed during the final step
+        """
+        # Intentionally do not process the auth form on the final step. We
+        # haven't stored this data, and it isn't required to login the user
+        form_list = self.get_form_list()
+        form_list.pop('auth')
+        return form_list
+
+    def process_step(self, form):
+        """
+        Process an individual step in the flow
+        """
+        # To prevent saving any private auth data to the session store, we
+        # validate the authentication form, determine the resulting user, then
+        # only store the minimum needed to login that user (the user's primary
+        # key and the backend used)
+        if self.steps.current == 'auth':
+            user = form.is_valid() and form.user_cache
+            self.storage.reset()
+            self.storage.authenticated_user = user
+            self.storage.data["authentication_time"] = int(time.time())
+
+            # By returning None when the user clicks the "back" button to the
+            # auth step the form will be blank with validation warnings
+            return None
+
+        return super().process_step(form)
+
+    def process_step_files(self, form):
+        """
+        Process the files submitted from a specific test
+        """
+        if self.steps.current == 'auth':
+            return {}
+        return super().process_step_files(form)
+
+    def get_form(self, *args, **kwargs):
+        """
+        Returns the form for the step
+        """
+        form = super().get_form(*args, **kwargs)
+        if self.show_timeout_error:
+            form.cleaned_data = getattr(form, 'cleaned_data', {})
+            form.add_error(None, ValidationError(_('Your session has timed out. Please login again.')))
+        return form
+
     def get_device(self, step=None):
         """
         Returns the OTP device selected by the user, or his default device.
@@ -187,9 +260,7 @@ def get_user(self):
         if not a valid user; see also issue #65.
         """
         if not self.user_cache:
-            form_obj = self.get_form(step='auth',
-                                     data=self.storage.get_step_data('auth'))
-            self.user_cache = form_obj.is_valid() and form_obj.user_cache
+            self.user_cache = self.storage.authenticated_user
         return self.user_cache
 
     def get_context_data(self, form, **kwargs):
diff --git a/two_factor/views/utils.py b/two_factor/views/utils.py
@@ -1,5 +1,6 @@
 import logging
 
+from django.contrib.auth import load_backend
 from django.core.exceptions import SuspiciousOperation
 from django.utils.decorators import method_decorator
 from django.utils.translation import gettext as _
@@ -37,6 +38,33 @@ def _set_validated_step_data(self, validated_step_data):
                                    _set_validated_step_data)
 
 
+class LoginStorage(ExtraSessionStorage):
+    """
+    SessionStorage that includes the property 'authenticated_user' for storing
+    backend authenticated users while logging in.
+    """
+    def _get_authenticated_user(self):
+        # Ensure that both user_pk and user_backend exist in the session
+        if not all([self.data.get("user_pk"), self.data.get("user_backend")]):
+            return False
+        # Acquire the user the same way django.contrib.auth.get_user does
+        backend = load_backend(self.data["user_backend"])
+        user = backend.get_user(self.data["user_pk"])
+        if not user:
+            return False
+        # Set user.backend to the dotted path version of the backend for login()
+        user.backend = self.data["user_backend"]
+        return user
+
+    def _set_authenticated_user(self, user):
+        # Acquire the PK the same way django's auth middleware does
+        self.data["user_pk"] = user._meta.pk.value_to_string(user)
+        self.data["user_backend"] = user.backend
+
+    authenticated_user = property(_get_authenticated_user,
+                                  _set_authenticated_user)
+
+
 class IdempotentSessionWizardView(SessionWizardView):
     """
     WizardView that allows certain steps to be marked non-idempotent, in which
@@ -153,6 +181,9 @@ def process_step(self, form):
 
         return super().process_step(form)
 
+    def get_done_form_list(self):
+        return self.get_form_list()
+
     def render_done(self, form, **kwargs):
         """
         This method gets called when all forms passed. The method should also
@@ -162,7 +193,7 @@ def render_done(self, form, **kwargs):
         """
         final_form_list = []
         # walk through the form list and try to validate the data again.
-        for form_key in self.get_form_list():
+        for form_key in self.get_done_form_list():
             form_obj = self.get_form(step=form_key,
                                      data=self.storage.get_step_data(form_key),
                                      files=self.storage.get_step_files(
