Merge pull request #6460 from lsd-cat/develop

Fix issues with token reuse:

securedrop/models.py

@@ -654,12 +654,6 @@ def verify_token(self, token: 'Optional[str]') -> bool:
         if not token:
             return False
 
-        token = self._format_token(token)
-
-        # Store latest token to prevent OTP token reuse
-        self.last_token = token
-        db.session.commit()
-
         if self.is_totp:
             # Also check the given token against the previous and next
             # valid tokens, to compensate for potential time skew
@@ -717,6 +711,9 @@ def login(cls,
         if len(user.otp_secret) < OTP_SECRET_MIN_ASCII_LENGTH:
             raise InvalidOTPSecretException(gettext("Invalid OTP secret"))
 
+        # From here to the return the order of statements is very important
+        token = user._format_token(token)
+
         if LOGIN_HARDENING:
             cls.throttle_login(user)
 
@@ -729,8 +726,14 @@ def login(cls,
                                             "{}".format(token))
         if not user.verify_token(token):
             raise BadTokenException("invalid two-factor code")
+
+        # Store latest token to prevent OTP token reuse
+        user.last_token = token
+        db.session.commit()
+
         if not user.valid_password(password):
             raise WrongPasswordException("invalid password")
+
         return user
 
     def generate_api_token(self, expiration: int) -> str:

securedrop/tests/test_2fa.py

@@ -44,7 +44,10 @@ def totp_window():
 
 
 def test_totp_reuse_protections(journalist_app, test_journo, hardening):
-    """Ensure that logging in twice with the same TOTP token fails."""
+    """Ensure that logging in twice with the same TOTP token fails.
+       Also, ensure that last_token is updated accordingly.
+    """
+
     with totp_window():
         token = TOTP(test_journo['otp_secret']).now()
 
@@ -53,11 +56,16 @@ def test_totp_reuse_protections(journalist_app, test_journo, hardening):
             resp = app.get(url_for('main.logout'), follow_redirects=True)
             assert resp.status_code == 200
 
+        with journalist_app.app_context():
+            journo = Journalist.query.get(test_journo['id'])
+            assert journo.last_token == token
+
         with journalist_app.test_client() as app:
             resp = app.post(url_for('main.login'),
                             data=dict(username=test_journo['username'],
                                       password=test_journo['password'],
                                       token=token))
+
             assert resp.status_code == 200
             text = resp.data.decode('utf-8')
             assert "Login failed" in text
@@ -81,6 +89,46 @@ def test_totp_reuse_protections2(journalist_app, test_journo, hardening):
                                  token)
 
 
+def test_totp_reuse_protections3(journalist_app, test_journo, hardening):
+    """We want to ensure that padding has no effect on token reuse verification."""
+
+    with totp_window():
+        token = TOTP(test_journo['otp_secret']).now()
+
+        with journalist_app.app_context():
+            Journalist.login(test_journo['username'],
+                             test_journo['password'],
+                             token)
+            with pytest.raises(BadTokenException):
+                Journalist.login(test_journo['username'],
+                                 test_journo['password'],
+                                 token + " ")
+
+
+def test_totp_reuse_protections4(journalist_app, test_journo, hardening):
+    """More granular than the preceeding test, we want to make sure the right
+       exception is being raised in the right place.
+    """
+
+    invalid_token = '000000'
+
+    with totp_window():
+        token = TOTP(test_journo['otp_secret']).now()
+
+        with journalist_app.app_context():
+            Journalist.login(test_journo['username'],
+                             test_journo['password'],
+                             token)
+            with pytest.raises(BadTokenException):
+                Journalist.login(test_journo['username'],
+                                 test_journo['password'],
+                                 invalid_token)
+            with pytest.raises(BadTokenException):
+                Journalist.login(test_journo['username'],
+                                 test_journo['password'],
+                                 token)
+
+
 def test_bad_token_fails_to_verify_on_admin_new_user_two_factor_page(
         journalist_app, test_admin, hardening):
     '''Regression test for
@@ -103,11 +151,6 @@ def test_bad_token_fails_to_verify_on_admin_new_user_two_factor_page(
                     'There was a problem verifying the two-factor code. Please try again.',
                     'error')
 
-        # last_token should be set to the token we just tried to use
-        with journalist_app.app_context():
-            admin = Journalist.query.get(test_admin['id'])
-            assert admin.last_token == invalid_token
-
         with journalist_app.test_client() as app:
             login_user(app, test_admin)
             # Submit the same invalid token again
@@ -142,11 +185,6 @@ def test_bad_token_fails_to_verify_on_new_user_two_factor_page(
                     'error'
                 )
 
-        # last_token should be set to the token we just tried to use
-        with journalist_app.app_context():
-            journo = Journalist.query.get(test_journo['id'])
-            assert journo.last_token == invalid_token
-
         with journalist_app.test_client() as app:
             login_user(app, test_journo)
 

securedrop/tests/test_journalist.py

@@ -767,7 +767,7 @@ def test_user_edits_password_error_response(config, journalist_app, test_journo,
             # to the database and this isolates the one we want to fail
             with patch.object(Journalist, 'verify_token', return_value=True):
                 with patch.object(db.session, 'commit',
-                                  side_effect=Exception()):
+                                  side_effect=[None, Exception()]):
                     with InstrumentedApp(journalist_app) as ins:
                         resp = app.post(
                             url_for('account.new_password', l=locale),