fix(accounts): OTP login works per user and "Lost OTP?" cannot impose 2FA

With OTP_ENABLED the login used django-otp's OTPAuthenticationForm. Under the
pinned django-otp 1.7.3 that form could not log anyone in through the shipped
page: the page posts only a 6-digit otp_token, but the form now needs the
device named explicitly, so an enrolled user got "Please select a device.",
and a user who never enrolled got "Please enter your OTP token." with no
device to produce one. Verified in an isolated test database.

SingleDeviceOTPForm makes OTP per user: a user with a confirmed TOTP device
must enter a valid code; a user without one logs in with the password alone.
The form chooses the device itself (the user's oldest confirmed TOTP device,
locked with select_for_update inside clean_otp's transaction so replay and
throttling counters cannot race) and ignores any posted otp_device, which
django-otp does not check for confirmed status or type. The login page drops
the HTML "required" on the token field (the field is optional for an
unenrolled user) and says so.

The "Lost OTP?" page (email_otp) created a confirmed device when none existed.
With per-user OTP that let anyone who knows an address turn a password-only
account into an OTP-required one and, if the mail never arrived, lock it out.
It now resends the QR only to a user who already enrolled and never creates a
device (get_existing_totp_device); admin enrollment (admin_email_otp) still
creates one. Email QR distribution and confirmed-on-create enrollment are
unchanged by user decision.

Tests: per-user login (no device -> password only; device -> valid token
passes, wrong/empty token fails), a posted or another user's device is not
used, a token cannot be replayed, and the real login page over HTTP with
OTP_ENABLED; email_otp sends nothing and creates no device for an unenrolled
user. The new tests fail on the previous code.

accounts/otp.py

@@ -0,0 +1,36 @@
+"""The login form used when OTP_ENABLED.
+
+django-otp's ``OTPAuthenticationForm`` requires every user to present a token
+and, since 1.6, to name the device explicitly, so the single-field login page
+cannot satisfy it and a user who never enrolled is locked out. This form makes
+OTP per user: a user with a confirmed TOTP device must enter a valid code, one
+without a device logs in with the password alone. The device is chosen by the
+server, so the page needs only the token field and a posted device id (which
+django-otp does not check for confirmed status or ownership-by-type) is
+ignored.
+"""
+
+from django_otp.forms import OTPAuthenticationForm
+from django_otp.plugins.otp_totp.models import TOTPDevice
+
+
+class SingleDeviceOTPForm(OTPAuthenticationForm):
+    def _confirmed_totp(self, user, for_verify=False):
+        """The user's oldest confirmed TOTP device, or None. With for_verify
+        the row is locked (select_for_update) so the token's replay and
+        throttling counters cannot race; call it only inside clean_otp's
+        transaction."""
+        devices = TOTPDevice.objects.filter(user=user, confirmed=True)
+        if for_verify:
+            devices = devices.select_for_update()
+        return devices.order_by("id").first()
+
+    def _chosen_device(self, user):
+        # Ignore a posted otp_device: it is verified within clean_otp's
+        # transaction, where the lock belongs.
+        return self._confirmed_totp(user, for_verify=True)
+
+    def clean_otp(self, user):
+        if user is not None and self._confirmed_totp(user) is None:
+            return  # no confirmed device: the password is the only factor
+        return super().clean_otp(user)

accounts/templates/accounts/otp_login.html

@@ -55,8 +55,9 @@
                 </div>
                 <div class="signin-input-group">
                     <span class="signin-input-icon"><i class="bi bi-key"></i></span>
-                    <input type="text" id="id_otp_token" class="signin-input form-control font-monospace" name="otp_token" placeholder="000000" inputmode="numeric" pattern="[0-9]*" maxlength="6" autofocus required>
+                    <input type="text" id="id_otp_token" class="signin-input form-control font-monospace" name="otp_token" placeholder="000000" inputmode="numeric" pattern="[0-9]*" maxlength="6" autofocus>
                 </div>
+                <small class="text-muted">{% trans "Leave empty if you have not set up two-factor authentication." %}</small>
             </div>
 
             <button id="btn-signin" class="btn btn-primary btn-signin-submit" type="submit">

accounts/test_email_otp.py

@@ -1,8 +1,10 @@
 """The "Lost OTP?" page mails a user their OTP QR code without a login. It
-exists only while OTP is enabled, mails one exact match at most once per
-interval, and never fails with an error page."""
+exists only while OTP is enabled; it resends the QR only to a user who
+has already enrolled (it never creates a device), mails one exact match at
+most once per interval, and never fails with an error page."""
 
 from django.contrib.auth import get_user_model
+from django_otp.plugins.otp_totp.models import TOTPDevice
 from django.core import mail
 from django.core.cache import cache
 from django.test import TestCase, override_settings
@@ -21,16 +23,26 @@ class EmailOTPTests(TestCase):
         cache.clear()
         self.user = get_user_model().objects.create_user("otp-user", email="otp.user@example.com", password="x")
 
+    def enroll(self, user=None):
+        return TOTPDevice.objects.create(user=user or self.user, confirmed=True)
+
     def post(self, email):
         response = self.client.post(URL, {"email": email})
         self.assertRedirects(response, reverse("accounts:login"), fetch_redirect_response=False)
         return response
 
-    def test_one_mail_to_the_user(self):
+    def test_one_mail_to_an_enrolled_user(self):
+        self.enroll()
         self.post("OTP.User@Example.com")  # addresses match case-insensitively
         self.assertEqual([m.to for m in mail.outbox], [["otp.user@example.com"]])
 
+    def test_an_unenrolled_user_gets_no_mail_and_no_device(self):
+        self.post("otp.user@example.com")
+        self.assertEqual(mail.outbox, [])
+        self.assertFalse(TOTPDevice.objects.filter(user=self.user).exists())
+
     def test_repeated_requests_send_one_mail_per_interval(self):
+        self.enroll()
         for _ in range(20):
             self.post("otp.user@example.com")
         self.assertEqual(len(mail.outbox), 1)
@@ -48,6 +60,7 @@ class EmailOTPTests(TestCase):
 
     @override_settings(EMAIL_BACKEND="django.core.mail.backends.smtp.EmailBackend", EMAIL_HOST="127.0.0.1", EMAIL_PORT=1)
     def test_a_mail_server_that_is_down_is_no_error_page(self):
+        self.enroll()
         with self.assertLogs("accounts.views", level="ERROR"):
             self.post("otp.user@example.com")
 

accounts/test_otp_login.py

@@ -0,0 +1,143 @@
+"""OTP login (SingleDeviceOTPForm): a user with a confirmed TOTP device must
+enter a valid code; a user without one logs in with the password alone. The
+device is chosen by the server (a posted device id is ignored), so the single
+token field of the login page works and no one else's device can be used."""
+
+from django.contrib.auth import get_user_model
+from django.test import RequestFactory, TestCase
+from django_otp.oath import totp
+from django_otp.plugins.otp_totp.models import TOTPDevice
+
+from accounts.otp import SingleDeviceOTPForm
+
+
+def token_for(device, offset=0):
+    return format(
+        totp(device.bin_key, step=device.step, t0=device.t0, digits=device.digits, drift=offset),
+        "0{}d".format(device.digits),
+    )
+
+
+def wrong_token(device):
+    """A code that is invalid now: it differs from the tokens of the current
+    step and its neighbours (which TOTP accepts within its tolerance), so the
+    test never flakes on a code that happens to match."""
+    valid = {token_for(device, d) for d in (-1, 0, 1)}
+    return next("%06d" % i for i in range(1000000) if "%06d" % i not in valid)
+
+
+class SingleDeviceOTPFormTestCase(TestCase):
+    def setUp(self):
+        self.User = get_user_model()
+        self.user = self.User.objects.create_user("alice", password="pw-correct-horse")
+
+    def _form(self, **extra):
+        data = {"username": "alice", "password": "pw-correct-horse", **extra}
+        return SingleDeviceOTPForm(RequestFactory().post("/accounts/login/"), data=data)
+
+    def _device(self, user=None, confirmed=True):
+        return TOTPDevice.objects.create(user=user or self.user, confirmed=confirmed)
+
+    def test_user_without_device_logs_in_with_password_only(self):
+        self.assertTrue(self._form(otp_token="").is_valid())
+
+    def test_user_with_device_needs_a_token(self):
+        self._device()
+        form = self._form(otp_token="")
+        self.assertFalse(form.is_valid())
+
+    def test_user_with_device_valid_token(self):
+        device = self._device()
+        self.assertTrue(self._form(otp_token=token_for(device)).is_valid())
+
+    def test_user_with_device_invalid_token(self):
+        device = self._device()
+        self.assertFalse(self._form(otp_token=wrong_token(device)).is_valid())
+
+    def test_single_token_field_without_device_id(self):
+        # the login page posts only otp_token; the form must still find the device
+        device = self._device()
+        form = self._form(otp_token=token_for(device))
+        self.assertTrue(form.is_valid(), form.errors)
+
+    def test_an_unconfirmed_device_does_not_require_otp(self):
+        self._device(confirmed=False)
+        self.assertTrue(self._form(otp_token="").is_valid())
+
+    def test_another_users_device_is_not_used(self):
+        other = self.User.objects.create_user("mallory", password="x")
+        other_device = self._device(user=other)
+        # alice has no device of her own: a token from mallory's device is ignored
+        self.assertTrue(self._form(otp_token=token_for(other_device)).is_valid())
+
+    def test_a_posted_device_id_is_ignored(self):
+        other = self.User.objects.create_user("mallory", password="x")
+        alice_device = self._device()
+        other_device = self._device(user=other)
+        form = self._form(otp_token=token_for(alice_device), otp_device=other_device.persistent_id)
+        self.assertTrue(form.is_valid(), form.errors)
+
+    def test_token_cannot_be_replayed(self):
+        device = self._device()
+        code = token_for(device)
+        self.assertTrue(self._form(otp_token=code).is_valid())
+        self.assertFalse(self._form(otp_token=code).is_valid())
+
+
+class OTPLoginHTTPTestCase(TestCase):
+    """The actual login page (OTP_ENABLED): the single token field logs an
+    enrolled user in, and a user without a device needs only the password.
+    Reloads the URLconf because accounts/urls.py picks the login form from
+    OTP_ENABLED at import time."""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        import importlib
+        from django.conf import settings
+        from django.urls import clear_url_caches
+        import accounts.urls
+        import webvirtcloud.urls
+
+        cls._otp_was_enabled = settings.OTP_ENABLED
+        settings.OTP_ENABLED = True
+        importlib.reload(accounts.urls)
+        importlib.reload(webvirtcloud.urls)
+        clear_url_caches()
+
+    @classmethod
+    def tearDownClass(cls):
+        import importlib
+        from django.conf import settings
+        from django.urls import clear_url_caches
+        import accounts.urls
+        import webvirtcloud.urls
+
+        settings.OTP_ENABLED = cls._otp_was_enabled
+        importlib.reload(accounts.urls)
+        importlib.reload(webvirtcloud.urls)
+        clear_url_caches()
+        super().tearDownClass()
+
+    def setUp(self):
+        self.user = get_user_model().objects.create_user("bob", password="pw-correct-horse")
+
+    def _post(self, **extra):
+        from django.urls import reverse
+        return self.client.post(reverse("accounts:login"), {
+            "username": "bob", "password": "pw-correct-horse", **extra,
+        })
+
+    def test_user_without_device_logs_in(self):
+        response = self._post(otp_token="")
+        self.assertEqual(response.status_code, 302)
+
+    def test_enrolled_user_logs_in_with_only_the_token(self):
+        device = TOTPDevice.objects.create(user=self.user, confirmed=True)
+        response = self._post(otp_token=token_for(device))
+        self.assertEqual(response.status_code, 302)
+
+    def test_enrolled_user_wrong_token_stays(self):
+        device = TOTPDevice.objects.create(user=self.user, confirmed=True)
+        response = self._post(otp_token=wrong_token(device))
+        self.assertEqual(response.status_code, 200)

accounts/test_utils.py

@@ -6,7 +6,7 @@ from django.core import mail
 from django.test import TestCase
 from django_otp.plugins.otp_totp.models import TOTPDevice
 
-from accounts.utils import get_user_totp_device, send_email_with_otp, validate_ssh_key
+from accounts.utils import (get_existing_totp_device, get_user_totp_device, send_email_with_otp, validate_ssh_key)
 
 
 class AccountsUtilsTestCase(TestCase):
@@ -60,6 +60,18 @@ class AccountsUtilsTestCase(TestCase):
         self.assertEqual(device1.id, device2.id)
         self.assertEqual(TOTPDevice.objects.filter(user=self.user).count(), 1)
 
+    def test_get_existing_totp_device_returns_none_without_a_device(self):
+        self.assertIsNone(get_existing_totp_device(self.user))
+        self.assertEqual(TOTPDevice.objects.filter(user=self.user).count(), 0)
+
+    def test_get_existing_totp_device_ignores_unconfirmed(self):
+        TOTPDevice.objects.create(user=self.user, confirmed=False)
+        self.assertIsNone(get_existing_totp_device(self.user))
+
+    def test_get_existing_totp_device_returns_confirmed(self):
+        device = TOTPDevice.objects.create(user=self.user, confirmed=True)
+        self.assertEqual(get_existing_totp_device(self.user).id, device.id)
+
     def test_send_email_with_otp(self):
         device = get_user_totp_device(self.user)
         send_email_with_otp(self.user, device)

accounts/urls.py

@@ -1,9 +1,9 @@
 from django.conf import settings
 from django.contrib.auth.views import LoginView, LogoutView
 from django.urls import path
-from django_otp.forms import OTPAuthenticationForm
 
 from . import views
+from .otp import SingleDeviceOTPForm
 from .views import CustomLoginView
 
 app_name = "accounts"
@@ -44,7 +44,7 @@ if settings.OTP_ENABLED:
             "login/",
             LoginView.as_view(
                 template_name="accounts/otp_login.html",
-                authentication_form=OTPAuthenticationForm,
+                authentication_form=SingleDeviceOTPForm,
             ),
             name="login",
         ),

accounts/utils.py

@@ -9,6 +9,16 @@ from django_otp import devices_for_user
 from django_otp.plugins.otp_totp.models import TOTPDevice
 
 
+def get_existing_totp_device(user):
+    """The user's confirmed TOTP device, or None. Unlike get_user_totp_device
+    it never creates one, so an unauthenticated caller cannot turn a
+    password-only account into an OTP-required one."""
+    for device in devices_for_user(user, confirmed=True):
+        if isinstance(device, TOTPDevice):
+            return device
+    return None
+
+
 def get_user_totp_device(user):
     devices = devices_for_user(user)
     for device in devices:

accounts/views.py

@@ -24,7 +24,7 @@ from accounts.forms import EmailOTPForm, ProfileForm, UserSSHKeyForm
 from accounts.models import User, UserAttributes, UserInstance, UserSSHKey
 
 from . import forms
-from .utils import get_user_totp_device, send_email_with_otp
+from .utils import get_existing_totp_device, get_user_totp_device, send_email_with_otp
 from django.contrib.auth.views import LoginView
 from logs.views import addlogmsg
 
@@ -221,10 +221,14 @@ def email_otp(request):
         # the cache is per process: with several workers the bound is per worker
         if len(users) == 1 and cache.add(f"email_otp:{users[0].pk}", True, EMAIL_OTP_INTERVAL):
             user = users[0]
-            try:
-                send_email_with_otp(user, get_user_totp_device(user))
-            except Exception:
-                logging.getLogger(__name__).exception("Could not mail the OTP QR code to user %s", user.pk)
+            # resend only to a user who already enrolled: creating a device here
+            # would let anyone who knows an email force OTP on that account
+            device = get_existing_totp_device(user)
+            if device is not None:
+                try:
+                    send_email_with_otp(user, device)
+                except Exception:
+                    logging.getLogger(__name__).exception("Could not mail the OTP QR code to user %s", user.pk)
 
         messages.success(
             request, _("OTP Sent to %(email)s") % {"email": form.cleaned_data["email"]}