diff --git a/accounts/otp.py b/accounts/otp.py
new file mode 100644
index 00000000..585f2b30
--- /dev/null
+++ b/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)
diff --git a/accounts/templates/accounts/otp_login.html b/accounts/templates/accounts/otp_login.html
index fcd22e96..a11fd4e0 100644
--- a/accounts/templates/accounts/otp_login.html
+++ b/accounts/templates/accounts/otp_login.html
@@ -55,8 +55,9 @@
+ {% trans "Leave empty if you have not set up two-factor authentication." %}
diff --git a/accounts/test_email_otp.py b/accounts/test_email_otp.py
index d6001eff..80d8e7ce 100644
--- a/accounts/test_email_otp.py
+++ b/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")
diff --git a/accounts/test_otp_login.py b/accounts/test_otp_login.py
new file mode 100644
index 00000000..98f3cdd3
--- /dev/null
+++ b/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)
diff --git a/accounts/test_utils.py b/accounts/test_utils.py
index 3af58b69..08d3b260 100644
--- a/accounts/test_utils.py
+++ b/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)
diff --git a/accounts/urls.py b/accounts/urls.py
index 041f6927..306741ac 100644
--- a/accounts/urls.py
+++ b/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",
),
diff --git a/accounts/utils.py b/accounts/utils.py
index a07a39b0..80fe2ec9 100644
--- a/accounts/utils.py
+++ b/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:
diff --git a/accounts/views.py b/accounts/views.py
index f5d6feac..80a539ee 100644
--- a/accounts/views.py
+++ b/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"]}