From 45828928abbee6e2f3a880e0127a9a6a64b993a5 Mon Sep 17 00:00:00 2001 From: catborise Date: Thu, 5 Nov 2026 18:44:14 +0300 Subject: [PATCH] fix(accounts): the admin's Email OTP sent twice at once makes one device get_user_totp_device looked for the user's TOTP device and made one if there was none; two requests at once could both make one, and both devices' codes then logged in. It now does so under a lock on the user's row. --- accounts/tests/test_utils.py | 8 ++++++++ accounts/utils.py | 18 +++++++++++------- 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/accounts/tests/test_utils.py b/accounts/tests/test_utils.py index 1c436aa5..bd84ec03 100644 --- a/accounts/tests/test_utils.py +++ b/accounts/tests/test_utils.py @@ -82,6 +82,14 @@ class AccountsUtilsTestCase(TestCase): device = TOTPDevice.objects.create(user=self.user, confirmed=True) self.assertEqual(get_existing_totp_device(self.user).id, device.id) + def test_the_device_is_made_under_a_lock_on_the_user(self): + from django.db.models.query import QuerySet + + with patch.object(QuerySet, "select_for_update", autospec=True, side_effect=QuerySet.select_for_update) as lock: + device = get_user_totp_device(self.user) + lock.assert_called_once() + self.assertEqual(get_user_totp_device(self.user), device) + @override_settings(OTP_ENABLED=True) def test_the_mail_links_to_the_qr_code_and_holds_no_secret(self): device = get_user_totp_device(self.user) diff --git a/accounts/utils.py b/accounts/utils.py index 80bdd9e2..a65ae96c 100644 --- a/accounts/utils.py +++ b/accounts/utils.py @@ -4,8 +4,10 @@ import struct from decimal import Decimal import segno +from django.contrib.auth import get_user_model from django.core.mail import send_mail from django.core.signing import TimestampSigner +from django.db import transaction from django.urls import reverse from django.utils.safestring import mark_safe from django.utils.translation import gettext as _ @@ -24,13 +26,15 @@ def get_existing_totp_device(user): def get_user_totp_device(user): - devices = devices_for_user(user) - for device in devices: - if isinstance(device, TOTPDevice): - return device - - device = user.totpdevice_set.create() - return device + """The user's TOTP device, made if they have none. The user's row is + locked meanwhile: two requests at once (a double click) make one device, + not two that both log in.""" + with transaction.atomic(): + get_user_model().objects.select_for_update().get(pk=user.pk) + for device in devices_for_user(user): + if isinstance(device, TOTPDevice): + return device + return user.totpdevice_set.create() def validate_ssh_key(key):