Prevent replay attacks for TOTP logins (baserow#4298)

Author: stribny

backend/src/baserow/core/migrations/0110_totpusedcode.py

@@ -0,0 +1,49 @@
+# Generated by Django 5.0.14 on 2026-01-07 10:53
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0109_userfile_deleted_at"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="TOTPUsedCode",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("used_at", models.DateTimeField()),
+                (
+                    "code",
+                    models.CharField(help_text="Hash of the used code", max_length=64),
+                ),
+                (
+                    "user",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="totp_used_codes",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "indexes": [
+                    models.Index(
+                        fields=["user", "code"], name="totp_usedcode_user_code_idx"
+                    )
+                ],
+            },
+        ),
+    ]

backend/src/baserow/core/two_factor_auth/models.py

@@ -57,6 +57,24 @@ def is_enabled(self):
         return self.enabled
 
 
+class TOTPUsedCode(models.Model):
+    user = models.ForeignKey(
+        "auth.User",
+        on_delete=models.CASCADE,
+        related_name="totp_used_codes",
+    )
+    used_at = models.DateTimeField()
+    code = models.CharField(max_length=64, help_text="Hash of the used code")
+
+    class Meta:
+        indexes = [
+            models.Index(
+                fields=["user", "code"],
+                name="totp_usedcode_user_code_idx",
+            ),
+        ]
+
+
 class TwoFactorAuthRecoveryCode(models.Model):
     user = models.ForeignKey(
         "auth.User",

backend/src/baserow/core/two_factor_auth/registries.py

@@ -26,6 +26,7 @@
 )
 from baserow.core.two_factor_auth.models import (
     TOTPAuthProviderModel,
+    TOTPUsedCode,
     TwoFactorAuthProviderModel,
     TwoFactorAuthRecoveryCode,
 )
@@ -134,6 +135,12 @@ def configure(
                 backup_codes_plaintext = self.generate_backup_codes()
                 self.store_backup_codes(provider, backup_codes_plaintext)
 
+                TOTPUsedCode.objects.create(
+                    user=provider.user,
+                    used_at=datetime.now(tz=timezone.utc),
+                    code=hashlib.sha256(code.encode("utf-8")).hexdigest(),
+                )
+
                 provider._backup_codes = backup_codes_plaintext
                 return provider
             else:
@@ -210,19 +217,36 @@ def verify(self, **kwargs) -> bool:
                 recovery_code.delete()
                 return True
 
-        provider = TwoFactorAuthProviderModel.objects.filter(user__email=email).first()
+        provider = (
+            TwoFactorAuthProviderModel.objects.select_for_update(of=("self",))
+            .filter(user__email=email)
+            .first()
+        )
         if not provider:
             raise VerificationFailed
 
         totp = pyotp.TOTP(provider.specific.secret)
+        current_ts = datetime.now(tz=timezone.utc)
+        hashed_code = hashlib.sha256(code.encode("utf-8")).hexdigest()
+
+        code_already_used = TOTPUsedCode.objects.filter(
+            user=provider.user, code=hashed_code
+        ).exists()
 
-        if totp.verify(code):
+        if not code_already_used and totp.verify(code):
+            TOTPUsedCode.objects.filter(user=provider.user).delete()
+            TOTPUsedCode.objects.create(
+                user=provider.user,
+                used_at=current_ts,
+                code=hashed_code,
+            )
             return True
         else:
             raise VerificationFailed
 
     def disable(self, provider, user):
         TwoFactorAuthRecoveryCode.objects.filter(user=user).delete()
+        TOTPUsedCode.objects.filter(user=user).delete()
         provider.delete()
 
 

backend/tests/baserow/api/two_factor_auth/test_two_factor_views.py

@@ -4,6 +4,7 @@
 
 import pyotp
 import pytest
+from freezegun import freeze_time
 from rest_framework.status import (
     HTTP_200_OK,
     HTTP_204_NO_CONTENT,
@@ -435,53 +436,55 @@ def test_verify_totp_view_missing_email(api_client, data_fixture):
 @pytest.mark.django_db
 def test_verify_totp_code_view(api_client, data_fixture):
     user, token = data_fixture.create_user_and_token()
-    provider = data_fixture.configure_totp(user)
-
-    response = api_client.post(
-        reverse("api:user:token_auth"),
-        {"email": user.email, "password": "password"},
-        format="json",
-    )
-    response_json = response.json()
-    two_fa_token = response_json["token"]
-
-    totp = pyotp.TOTP(provider.secret)
-    valid_code = totp.now()
-
-    url = reverse("api:two_factor_auth:verify")
-    response = api_client.post(
-        url,
-        {
-            "type": "totp",
-            "email": user.email,
-            "code": valid_code,
-        },
-        format="json",
-        HTTP_AUTHORIZATION=f"Bearer {two_fa_token}",
-    )
-
-    response_json = response.json()
-    assert response.status_code == HTTP_200_OK, response_json
-    assert response_json == {
-        "access_token": AnyStr(),
-        "active_licenses": {"instance_wide": {}, "per_workspace": {}},
-        "permissions": AnyList(),
-        "refresh_token": AnyStr(),
-        "token": AnyStr(),
-        "user": {
-            "completed_guided_tours": [],
-            "completed_onboarding": False,
-            "email_notification_frequency": "instant",
-            "email_verified": False,
-            "first_name": user.first_name,
-            "id": user.id,
-            "is_staff": False,
-            "language": "en",
-            "username": user.email,
-        },
-        "user_notifications": {"unread_count": 0},
-        "user_session": AnyStr(),
-    }
+    with freeze_time("2020-02-01 00:00"):
+        provider = data_fixture.configure_totp(user)
+
+    with freeze_time("2020-02-01 00:01"):
+        response = api_client.post(
+            reverse("api:user:token_auth"),
+            {"email": user.email, "password": "password"},
+            format="json",
+        )
+        response_json = response.json()
+        two_fa_token = response_json["token"]
+
+        totp = pyotp.TOTP(provider.secret)
+        valid_code = totp.now()
+
+        url = reverse("api:two_factor_auth:verify")
+        response = api_client.post(
+            url,
+            {
+                "type": "totp",
+                "email": user.email,
+                "code": valid_code,
+            },
+            format="json",
+            HTTP_AUTHORIZATION=f"Bearer {two_fa_token}",
+        )
+
+        response_json = response.json()
+        assert response.status_code == HTTP_200_OK, response_json
+        assert response_json == {
+            "access_token": AnyStr(),
+            "active_licenses": {"instance_wide": {}, "per_workspace": {}},
+            "permissions": AnyList(),
+            "refresh_token": AnyStr(),
+            "token": AnyStr(),
+            "user": {
+                "completed_guided_tours": [],
+                "completed_onboarding": False,
+                "email_notification_frequency": "instant",
+                "email_verified": False,
+                "first_name": user.first_name,
+                "id": user.id,
+                "is_staff": False,
+                "language": "en",
+                "username": user.email,
+            },
+            "user_notifications": {"unread_count": 0},
+            "user_session": AnyStr(),
+        }
 
 
 @pytest.mark.django_db

backend/tests/baserow/core/two_factor_auth/test_two_factor_handler.py

@@ -2,6 +2,7 @@
 
 import pyotp
 import pytest
+from freezegun import freeze_time
 
 from baserow.core.two_factor_auth.exceptions import (
     TwoFactorAuthCannotBeConfigured,
@@ -144,9 +145,13 @@ def test_verify_no_provider(data_fixture):
 
 @pytest.mark.django_db
 def test_verify(data_fixture):
-    user = data_fixture.create_user(password="password")
-    provider = data_fixture.configure_totp(user)
-    totp = pyotp.TOTP(provider.secret)
-    code = totp.now()
-
-    assert TwoFactorAuthHandler().verify("totp", email=user.email, code=code) is True
+    with freeze_time("2020-02-01 00:00"):
+        user = data_fixture.create_user(password="password")
+        provider = data_fixture.configure_totp(user)
+        totp = pyotp.TOTP(provider.secret)
+
+    with freeze_time("2020-02-01 00:05"):
+        code = totp.now()
+        assert (
+            TwoFactorAuthHandler().verify("totp", email=user.email, code=code) is True
+        )

backend/tests/baserow/core/two_factor_auth/test_two_factor_registries.py

@@ -11,6 +11,7 @@
 )
 from baserow.core.two_factor_auth.models import (
     TOTPAuthProviderModel,
+    TOTPUsedCode,
     TwoFactorAuthRecoveryCode,
 )
 from baserow.core.two_factor_auth.registries import TOTPAuthProviderType
@@ -66,6 +67,7 @@ def test_totp_configure_finish_configuration(data_fixture):
     assert provider.provisioning_qr_code == ""
     assert TOTPAuthProviderModel.objects.filter(user=user).count() == 1
     assert TwoFactorAuthRecoveryCode.objects.filter(user=user).count() == 8
+    assert TOTPUsedCode.objects.filter(user=user).count() == 1
 
 
 @pytest.mark.django_db
@@ -123,22 +125,62 @@ def test_generate_backup_codes():
 @pytest.mark.django_db
 def test_verify_with_code(data_fixture):
     user = data_fixture.create_user()
-    provider = data_fixture.configure_totp(user)
+    with freeze_time("2020-02-01 00:00"):
+        provider = data_fixture.configure_totp(user)
+        assert TOTPUsedCode.objects.filter(user=user).count() == 1
+
     totp = pyotp.TOTP(provider.secret)
-    code = totp.now()
 
-    assert TOTPAuthProviderType().verify(email=user.email, code=code)
+    with freeze_time("2020-02-01 00:05"):
+        code = totp.now()
+        assert TOTPAuthProviderType().verify(email=user.email, code=code)
+        assert TOTPUsedCode.objects.filter(user=user).count() == 1
 
 
 @pytest.mark.django_db
-def test_verify_with_code_fails(data_fixture):
+def test_verify_with_code_fails_wrong_code(data_fixture):
     user = data_fixture.create_user()
     data_fixture.configure_totp(user)
 
     with pytest.raises(VerificationFailed):
         TOTPAuthProviderType().verify(email=user.email, code="1234567")
 
 
+@pytest.mark.django_db
+def test_verify_with_code_code_cannot_be_reused(data_fixture):
+    user = data_fixture.create_user()
+    user_2 = data_fixture.create_user()
+    with freeze_time("2020-02-01 00:00"):
+        provider = data_fixture.configure_totp(user)
+        provider_2 = data_fixture.configure_totp(user_2)
+        provider_2.secret = provider.secret
+        provider_2.save()
+    totp = pyotp.TOTP(provider.secret)
+
+    with freeze_time("2020-02-01 00:05"):
+        code = totp.now()
+
+        # first time the code is valid
+        assert TOTPAuthProviderType().verify(email=user.email, code=code)
+
+        with pytest.raises(VerificationFailed):
+            TOTPAuthProviderType().verify(email=user.email, code=code)
+
+        # another user with the same secret would not be
+        # affected
+        assert TOTPAuthProviderType().verify(email=user_2.email, code=code)
+
+    with freeze_time("2020-02-01 00:10"):
+        code = totp.now()
+
+        # user can use a new code from a different time window
+        assert TOTPAuthProviderType().verify(email=user.email, code=code)
+
+        # older used codes are automatically deleted upon successful
+        # verification
+        assert TOTPUsedCode.objects.filter(user=user).count() == 1
+
+
 @pytest.mark.django_db
 def test_verify_with_backup_code(data_fixture):
     user = data_fixture.create_user()
@@ -169,9 +211,14 @@ def test_verify_no_provider(data_fixture):
 @pytest.mark.django_db
 def test_totp_disable(data_fixture):
     user = data_fixture.create_user()
+    user_2 = data_fixture.create_user()
     provider = data_fixture.configure_totp(user)
+    data_fixture.configure_totp(user_2)
+    assert TOTPUsedCode.objects.count() == 2
 
     TOTPAuthProviderType().disable(provider, user)
 
     assert TOTPAuthProviderModel.objects.filter(user=user).count() == 0
     assert TwoFactorAuthRecoveryCode.objects.filter(user=user).count() == 0
+    assert TOTPUsedCode.objects.count() == 1
+    assert TOTPUsedCode.objects.filter(user=user_2).count() == 1