[pull] develop from baserow:develop

backend/src/baserow/api/two_factor_auth/serializers.py

@@ -35,6 +35,10 @@ class DisableTwoFactorAuthSerializer(serializers.Serializer):
 
 
 class VerifyTOTPSerializer(serializers.Serializer):
-    email = serializers.EmailField(required=True)
+    email = serializers.EmailField(
+        required=False,
+        help_text="Deprecated. User identity is resolved from the 2FA token. "
+        "This field is ignored and will be removed in a future version.",
+    )
     code = serializers.CharField(required=False)
     backup_code = serializers.CharField(required=False)

backend/src/baserow/api/two_factor_auth/tokens.py

@@ -1,29 +1,52 @@
 from django.conf import settings
+from django.contrib.auth import get_user_model
 
-from rest_framework import permissions
+from rest_framework import exceptions
+from rest_framework.authentication import BaseAuthentication
 from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
+from rest_framework_simplejwt.settings import api_settings as jwt_settings
 from rest_framework_simplejwt.tokens import Token
 
+User = get_user_model()
+
 
 class TwoFactorAccessToken(Token):
     token_type = "2fa"  # nosec
     lifetime = settings.ACCESS_TOKEN_LIFETIME
 
 
-class Require2faToken(permissions.BasePermission):
+class TwoFactorTokenAuthentication(BaseAuthentication):
     """
-    Require that the provided JWT is two factor access token type.
+    Authenticates the request by validating a 2FA JWT from the
+    Authorization header and resolving the user from the token payload.
     """
 
-    def has_permission(self, request, view):
+    def authenticate_header(self, request):
+        return "Bearer"
+
+    def authenticate(self, request):
         auth_header = request.headers.get("Authorization", "")
         if not auth_header.startswith("Bearer "):
-            return False
+            return None
 
         token_string = auth_header.split(" ")[1]
 
         try:
             token = TwoFactorAccessToken(token_string)
-            return token.token_type == "2fa"  # nosec
         except (InvalidToken, TokenError):
-            return False
+            raise exceptions.AuthenticationFailed("Invalid 2FA token.")
+
+        try:
+            user_id = token.payload[jwt_settings.USER_ID_CLAIM]
+        except KeyError:
+            raise exceptions.AuthenticationFailed(
+                "Token contained no recognizable user identification."
+            )
+
+        user = User.objects.filter(
+            **{jwt_settings.USER_ID_FIELD: user_id}, is_active=True
+        ).first()
+        if not user:
+            raise exceptions.AuthenticationFailed("User not found or inactive.")
+
+        return (user, token)

backend/src/baserow/api/two_factor_auth/views.py

@@ -27,11 +27,10 @@
     TwoFactorAuthSerializer,
     VerifyTOTPSerializer,
 )
-from baserow.api.two_factor_auth.tokens import Require2faToken
+from baserow.api.two_factor_auth.tokens import TwoFactorTokenAuthentication
 from baserow.api.user.schemas import authenticated_user_response_schema
 from baserow.api.user.serializers import log_in_user
 from baserow.api.utils import DiscriminatorCustomFieldsMappingSerializer
-from baserow.core.models import User
 from baserow.core.two_factor_auth.actions import (
     ConfigureTwoFactorAuthActionType,
     DisableTwoFactorAuthActionType,
@@ -177,7 +176,8 @@ def post(self, request, data: dict):
 
 
 class VerifyTOTPAuthView(APIView):
-    permission_classes = (Require2faToken,)
+    authentication_classes = (TwoFactorTokenAuthentication,)
+    permission_classes = (IsAuthenticated,)
 
     @extend_schema(
         tags=["Auth"],
@@ -210,16 +210,22 @@ def post(self, request, data: dict):
         Verifies TOTP two-factor authentication.
         """
 
+        user = request.user
+
         def verify():
-            TwoFactorAuthHandler().verify(TOTPAuthProviderType.type, **data)
+            TwoFactorAuthHandler().verify(
+                TOTPAuthProviderType.type,
+                email=user.email,
+                code=data.get("code"),
+                backup_code=data.get("backup_code"),
+            )
 
         rate_limit(
             rate=RateLimit.from_string("10/m"),
-            key=f"two_fa_verify:totp:{data.get('email', '')}",
+            key=f"two_fa_verify:totp:{user.id}",
             raise_exception=True,
         )(verify)()
 
-        user = User.objects.filter(email=data["email"]).first()
         return_data = log_in_user(request, user)
 
         return Response(

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

@@ -407,7 +407,12 @@ def test_disable_2fa_view(api_client, data_fixture):
 
 
 @pytest.mark.django_db
-def test_verify_totp_view_missing_email(api_client, data_fixture):
+def test_verify_totp_view_without_email(api_client, data_fixture):
+    """
+    The email field is deprecated and optional. Omitting it should not
+    cause a validation error — the request proceeds to TOTP verification.
+    """
+
     user, token = data_fixture.create_user_and_token()
     data_fixture.configure_totp(user)
     response = api_client.post(
@@ -429,8 +434,8 @@ def test_verify_totp_view_missing_email(api_client, data_fixture):
     )
 
     response_json = response.json()
-    assert response.status_code == HTTP_400_BAD_REQUEST, response_json
-    assert response_json["error"] == "ERROR_REQUEST_BODY_VALIDATION"
+    assert response.status_code == HTTP_401_UNAUTHORIZED, response_json
+    assert response_json["error"] == "ERROR_TWO_FACTOR_AUTH_VERIFICATION_FAILED"
 
 
 @pytest.mark.django_db
@@ -595,3 +600,206 @@ def test_verify_totp_backup_code_view_invalid(api_client, data_fixture):
     response_json = response.json()
     assert response.status_code == HTTP_401_UNAUTHORIZED, response_json
     assert response_json["error"] == "ERROR_TWO_FACTOR_AUTH_VERIFICATION_FAILED"
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("use_correct_email", [True, False])
+def test_verify_totp_uses_token_identity_not_body_email(
+    api_client, data_fixture, use_correct_email
+):
+    """
+    The verify endpoint resolves the user from the 2FA token, not from
+    the email field in the request body. A valid TOTP code for the token
+    owner succeeds regardless of what email is sent in the body.
+    """
+
+    user, _ = data_fixture.create_user_and_token()
+    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",
+        )
+        two_fa_token = response.json()["token"]
+
+        totp = pyotp.TOTP(provider.secret)
+        valid_code = totp.now()
+
+        body_email = user.email if use_correct_email else "wrong@example.com"
+        url = reverse("api:two_factor_auth:verify")
+        response = api_client.post(
+            url,
+            {
+                "type": "totp",
+                "email": body_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["user"]["id"] == user.id
+
+
+@pytest.mark.django_db
+def test_verify_totp_cross_account_splicing_blocked(api_client, data_fixture):
+    """
+    An attacker who obtains their own 2FA token must not be able to use it
+    with a victim's email and TOTP code to get the victim's session.
+    The endpoint resolves user from the token — so the attacker's TOTP
+    code is checked, not the victim's.
+    """
+
+    attacker, _ = data_fixture.create_user_and_token(email="attacker@test.com")
+    victim, _ = data_fixture.create_user_and_token(email="victim@test.com")
+
+    data_fixture.configure_totp(attacker)
+    victim_provider = data_fixture.configure_totp(victim)
+    victim_backup_code = victim_provider.backup_codes[0]
+
+    response = api_client.post(
+        reverse("api:user:token_auth"),
+        {"email": attacker.email, "password": "password"},
+        format="json",
+    )
+    attacker_2fa_token = response.json()["token"]
+
+    url = reverse("api:two_factor_auth:verify")
+    response = api_client.post(
+        url,
+        {
+            "type": "totp",
+            "email": victim.email,
+            "backup_code": victim_backup_code,
+        },
+        format="json",
+        HTTP_AUTHORIZATION=f"Bearer {attacker_2fa_token}",
+    )
+
+    # The endpoint uses the attacker's token identity, so the victim's
+    # backup code is checked against the attacker's codes and fails.
+    response_json = response.json()
+    assert response.status_code == HTTP_401_UNAUTHORIZED, response_json
+    assert response_json["error"] == "ERROR_TWO_FACTOR_AUTH_VERIFICATION_FAILED"
+
+
+@pytest.mark.django_db
+def test_verify_totp_deleted_user_in_token(api_client, data_fixture):
+    """
+    If the user referenced by the 2FA token has been deleted between token
+    issuance and verification, the authentication layer rejects the request.
+    """
+
+    user, _ = data_fixture.create_user_and_token()
+    data_fixture.configure_totp(user)
+
+    response = api_client.post(
+        reverse("api:user:token_auth"),
+        {"email": user.email, "password": "password"},
+        format="json",
+    )
+    two_fa_token = response.json()["token"]
+
+    user.delete()
+
+    url = reverse("api:two_factor_auth:verify")
+    response = api_client.post(
+        url,
+        {
+            "type": "totp",
+            "code": "123456",
+        },
+        format="json",
+        HTTP_AUTHORIZATION=f"Bearer {two_fa_token}",
+    )
+
+    assert response.status_code == HTTP_401_UNAUTHORIZED
+
+
+@pytest.mark.django_db
+def test_verify_totp_deactivated_user_in_token(api_client, data_fixture):
+    """
+    If the user referenced by the 2FA token has been deactivated between
+    token issuance and verification, the authentication layer rejects the
+    request.
+    """
+
+    user, _ = data_fixture.create_user_and_token()
+    data_fixture.configure_totp(user)
+
+    response = api_client.post(
+        reverse("api:user:token_auth"),
+        {"email": user.email, "password": "password"},
+        format="json",
+    )
+    two_fa_token = response.json()["token"]
+
+    user.is_active = False
+    user.save()
+
+    url = reverse("api:two_factor_auth:verify")
+    response = api_client.post(
+        url,
+        {
+            "type": "totp",
+            "code": "123456",
+        },
+        format="json",
+        HTTP_AUTHORIZATION=f"Bearer {two_fa_token}",
+    )
+
+    assert response.status_code == HTTP_401_UNAUTHORIZED
+
+
+@pytest.mark.django_db
+def test_verify_totp_rate_limit_keyed_to_token_user_id(api_client, data_fixture):
+    """
+    Rate limiting is keyed to the user_id from the token, not the email
+    in the request body. Sending different emails in the body should still
+    hit the same rate limit bucket when the same token is used.
+    """
+
+    user, _ = data_fixture.create_user_and_token()
+    data_fixture.configure_totp(user)
+
+    response = api_client.post(
+        reverse("api:user:token_auth"),
+        {"email": user.email, "password": "password"},
+        format="json",
+    )
+    two_fa_token = response.json()["token"]
+
+    url = reverse("api:two_factor_auth:verify")
+
+    for i in range(10):
+        response = api_client.post(
+            url,
+            {
+                "type": "totp",
+                "backup_code": "XXXXX-XXXXX",
+            },
+            format="json",
+            HTTP_AUTHORIZATION=f"Bearer {two_fa_token}",
+        )
+        assert response.status_code != 429, (
+            f"Request {i + 1} was rate limited prematurely"
+        )
+
+    # 11th request with same token should be rate limited
+    response = api_client.post(
+        url,
+        {
+            "type": "totp",
+            "backup_code": "XXXXX-XXXXX",
+        },
+        format="json",
+        HTTP_AUTHORIZATION=f"Bearer {two_fa_token}",
+    )
+
+    assert response.status_code == 429, response.json()
+    assert response.json()["error"] == "ERROR_RATE_LIMIT_EXCEEDED"

changelog/entries/unreleased/refactor/5743_simplify_2fa_verify_endpoint_to_resolve_user_identity_from_t.json

@@ -0,0 +1,9 @@
+{
+    "type": "refactor",
+    "message": "Simplify 2FA verify endpoint to resolve user identity from the authentication token instead of the request body.",
+    "issue_origin": "github",
+    "issue_number": 5743,
+    "domain": "core",
+    "bullet_points": [],
+    "created_at": "2026-07-20"
+}