add auth test for replay attacks

piccolo_api/mfa/authenticator/tables.py

@@ -65,33 +65,35 @@ async def create_new(cls, user_id: int) -> AuthenticatorSecret:
 
     @classmethod
     async def authenticate(cls, user_id: int, code: str) -> bool:
-        seeds = await cls.objects().where(
+        secrets = await cls.objects().where(
             cls.user_id == user_id,
             cls.revoked_at.is_null(),
         )
 
-        if not seeds:
+        if not secrets:
             return False
 
         pyotp = get_pyotp()
 
         # We check all seeds - a user is allowed multiple seeds (i.e. if they
         # have multiple devices).
-        for seed in seeds:
-            if seed.last_used_code == code:
+        for secret in secrets:
+            if secret.last_used_code == code:
                 logger.warning(
                     f"User {user_id} reused a token - potential replay attack."
                 )
                 return False
 
-            totp = pyotp.TOTP(seed.secret)
+            totp = pyotp.TOTP(secret.secret)
 
             if totp.verify(code):
-                seed.last_used_at = datetime.datetime.now(
+                secret.last_used_at = datetime.datetime.now(
                     tz=datetime.timezone.utc
                 )
-                seed.last_used_code = code
-                await seed.save(columns=[cls.last_used_at, cls.last_used_code])
+                secret.last_used_code = code
+                await secret.save(
+                    columns=[cls.last_used_at, cls.last_used_code]
+                )
 
                 return True
 

tests/mfa/authenticator/test_tables.py

@@ -1,5 +1,6 @@
 import datetime
 from unittest import TestCase
+from unittest.mock import MagicMock, patch
 
 from piccolo.apps.user.tables import BaseUser
 from piccolo.testing.test_case import AsyncTableTest
@@ -21,6 +22,36 @@ def test_generate_secret(self):
         self.assertEqual(len(secret_1), 32)
 
 
+class TestAuthenticate(AsyncTableTest):
+
+    tables = [AuthenticatorSecret, BaseUser]
+
+    @patch("piccolo_api.mfa.authenticator.tables.logger")
+    async def test_replay_attack(self, logger: MagicMock):
+        """
+        If a token which was just used successfully is reused, it should be
+        rejected, because it might be a replay attack.
+        """
+        user = await BaseUser.create_user(
+            username="test", password="test123456"
+        )
+
+        code = "123456"
+
+        seed = await AuthenticatorSecret.create_new(user_id=user.id)
+        seed.last_used_code = code
+        await seed.save()
+
+        auth_response = await AuthenticatorSecret.authenticate(
+            user_id=user.id, code=code
+        )
+        assert auth_response is False
+
+        logger.warning.assert_called_with(
+            "User 1 reused a token - potential replay attack."
+        )
+
+
 class TestCreateNew(AsyncTableTest):
 
     tables = [AuthenticatorSecret, BaseUser]