From a8a04545827405ed03099bd358471fe2971104a2 Mon Sep 17 00:00:00 2001 From: dimmur-brw Date: Tue, 21 Jul 2026 09:52:07 +0200 Subject: [PATCH] Simplify 2FA verify endpoint user resolution (#5744) --- .../api/two_factor_auth/serializers.py | 6 +- .../src/baserow/api/two_factor_auth/tokens.py | 37 ++- .../src/baserow/api/two_factor_auth/views.py | 18 +- .../two_factor_auth/test_two_factor_views.py | 214 +++++++++++++++++- ...point_to_resolve_user_identity_from_t.json | 9 + 5 files changed, 267 insertions(+), 17 deletions(-) create mode 100644 changelog/entries/unreleased/refactor/5743_simplify_2fa_verify_endpoint_to_resolve_user_identity_from_t.json diff --git a/backend/src/baserow/api/two_factor_auth/serializers.py b/backend/src/baserow/api/two_factor_auth/serializers.py index 32b287400e..b3021bbb5b 100644 --- a/backend/src/baserow/api/two_factor_auth/serializers.py +++ b/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) diff --git a/backend/src/baserow/api/two_factor_auth/tokens.py b/backend/src/baserow/api/two_factor_auth/tokens.py index 5559cf66e4..1c6e11974e 100644 --- a/backend/src/baserow/api/two_factor_auth/tokens.py +++ b/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) diff --git a/backend/src/baserow/api/two_factor_auth/views.py b/backend/src/baserow/api/two_factor_auth/views.py index ad85e282a2..0ff597ad5f 100644 --- a/backend/src/baserow/api/two_factor_auth/views.py +++ b/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( diff --git a/backend/tests/baserow/api/two_factor_auth/test_two_factor_views.py b/backend/tests/baserow/api/two_factor_auth/test_two_factor_views.py index fbf46d8432..d8dc773459 100644 --- a/backend/tests/baserow/api/two_factor_auth/test_two_factor_views.py +++ b/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" diff --git a/changelog/entries/unreleased/refactor/5743_simplify_2fa_verify_endpoint_to_resolve_user_identity_from_t.json b/changelog/entries/unreleased/refactor/5743_simplify_2fa_verify_endpoint_to_resolve_user_identity_from_t.json new file mode 100644 index 0000000000..db4cc95b61 --- /dev/null +++ b/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" +}