Passwordless users can also use elevate

hypha/apply/users/tests/test_email_change.py

@@ -25,6 +25,17 @@ def url_with_value(signed_value):
     return f"{EMAIL_CHANGE_URL}?{urlencode({'value': signed_value})}"
 
 
+class ElevatedSessionMixin:
+    """Treat the session as elevated for the duration of the test."""
+
+    def elevate_session(self):
+        patcher = patch(
+            "hypha.elevate.middleware.has_elevated_privileges", return_value=True
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+
 class TestEmailChangeRequiresLogin(TestCase):
     def test_unauthenticated_user_redirected_to_login(self):
         from django.conf import settings
@@ -35,7 +46,7 @@ def test_unauthenticated_user_redirected_to_login(self):
 
 
 class TestEmailChangeElevationCheck(TestCase):
-    """Users with a usable password must re-authenticate (elevate) before proceeding."""
+    """Every user must re-authenticate (elevate) before proceeding."""
 
     def setUp(self):
         self.user = UserFactory()  # has a usable password
@@ -62,25 +73,35 @@ def test_elevated_user_is_not_redirected_to_elevate(self):
         self.assertNotEqual(response["Location"], ELEVATE_URL)
 
 
-class TestEmailChangeOAuthUserSkipsElevation(TestCase):
-    """OAuth users have no usable password — the elevation gate must be skipped."""
+class TestEmailChangeOAuthUserRequiresElevation(TestCase):
+    """OAuth users have no usable password, but still have to confirm access.
+
+    The elevate page offers them an emailed confirmation code instead of a
+    password prompt.
+    """
 
     def setUp(self):
         self.user = OAuthUserFactory()
         self.client.force_login(self.user)
 
-    def test_oauth_user_not_redirected_to_elevate(self):
+    def test_oauth_user_redirected_to_elevate(self):
         signed = make_signed_value(self.user.email)
         response = self.client.get(url_with_value(signed), follow=False)
-        self.assertNotIn(ELEVATE_URL, response.get("Location", ""))
+        self.assertEqual(response.status_code, 302)
+        self.assertIn(ELEVATE_URL, response["Location"])
+
+    def test_elevate_page_offers_confirmation_code(self):
+        response = self.client.get(ELEVATE_URL, follow=False)
+        self.assertContains(response, "Send a confirmation code to your email")
 
 
-class TestEmailChangeTokenValidation(TestCase):
+class TestEmailChangeTokenValidation(ElevatedSessionMixin, TestCase):
     """The signed token in the query string must be valid and unexpired."""
 
     def setUp(self):
-        self.user = OAuthUserFactory()  # skip elevation
+        self.user = OAuthUserFactory()
         self.client.force_login(self.user)
+        self.elevate_session()
 
     def test_missing_value_param_redirects_to_account(self):
         response = self.client.get(EMAIL_CHANGE_URL, follow=False)
@@ -101,12 +122,13 @@ def test_tampered_value_shows_error_message(self):
         self.assertContains(response, "timed out")
 
 
-class TestEmailChangeSuccess(TestCase):
+class TestEmailChangeSuccess(ElevatedSessionMixin, TestCase):
     """With a valid elevated session and correct token, the view updates the user."""
 
     def setUp(self):
-        self.user = OAuthUserFactory()  # skip elevation
+        self.user = OAuthUserFactory()
         self.client.force_login(self.user)
+        self.elevate_session()
 
     def test_valid_token_redirects_to_confirm_link_sent(self):
         signed = make_signed_value(self.user.email, name="New Name")

hypha/apply/users/tests/test_ratelimit.py

@@ -1,6 +1,9 @@
 from django.test import TestCase
 from django.urls import reverse
+from django.utils.encoding import force_bytes
+from django.utils.http import urlsafe_base64_encode
 
+from ..tokens import PasswordlessLoginTokenGenerator
 from .factories import UserFactory
 
 LOGIN_URL = reverse("users:login")
@@ -11,16 +14,22 @@
 
 
 class TestLoginViewRateLimit(TestCase):
-    """Login view is protected by an IP-based rate limit on POST requests."""
+    """Login view is rate-limited by both IP and account.
 
-    def _post_login(self, email="test@example.com"):
+    The account key has to read `auth-username`: the view is a two-factor
+    wizard, so it prefixes its form fields and there is no plain `email` field
+    to key on.
+    """
+
+    def _post_login(self, email="test@example.com", ip="127.0.0.1"):
         return self.client.post(
             LOGIN_URL,
             data={
                 "login_view-current_step": "auth",
                 "auth-username": email,
                 "auth-password": "wrong-password",
             },
+            REMOTE_ADDR=ip,
         )
 
     def test_login_accessible_before_limit(self):
@@ -34,6 +43,58 @@ def test_login_blocked_after_ip_limit_exceeded(self):
         response = self._post_login()
         self.assertEqual(response.status_code, 403)
 
+    def test_login_blocked_after_account_limit_exceeded(self):
+        """Password-spraying one account is throttled across IPs."""
+        user = UserFactory()
+        for i in range(RATE_LIMIT):
+            self._post_login(email=user.email, ip=f"10.0.0.{i}")
+        response = self._post_login(email=user.email, ip="10.0.0.99")
+        self.assertEqual(response.status_code, 403)
+
+    def test_account_limit_does_not_lock_out_other_accounts(self):
+        """A per-account key must not collapse into one site-wide bucket.
+
+        If it does, anyone can exhaust the limit and block password login for
+        every user — an unauthenticated denial of service.
+        """
+        victim = UserFactory()
+        for i in range(RATE_LIMIT):
+            self._post_login(email=f"attacker{i}@example.com", ip=f"10.0.0.{i}")
+        response = self._post_login(email=victim.email, ip="10.0.0.99")
+        self.assertNotEqual(response.status_code, 403)
+
+    def test_username_key_is_case_and_whitespace_insensitive(self):
+        """Casing the address differently must not buy a fresh bucket."""
+        user = UserFactory(email="Victim@Example.com")
+        for i in range(RATE_LIMIT):
+            self._post_login(email=f" {user.email.upper()} ", ip=f"10.0.0.{i}")
+        response = self._post_login(email=user.email.lower(), ip="10.0.0.99")
+        self.assertEqual(response.status_code, 403)
+
+
+class TestPasswordlessLoginRateLimit(TestCase):
+    """`PasswordlessLoginView` inherits `LoginView`'s decorated `dispatch`.
+
+    Its POSTs carry no username, so they key on IP — one user clicking a magic
+    link must never consume a budget shared with everyone else's.
+    """
+
+    def _confirm_login(self, user, ip):
+        url = reverse(
+            "users:do_passwordless_login",
+            kwargs={
+                "uidb64": urlsafe_base64_encode(force_bytes(user.pk)),
+                "token": PasswordlessLoginTokenGenerator().make_token(user),
+            },
+        )
+        return self.client.post(url, REMOTE_ADDR=ip)
+
+    def test_one_users_confirmations_do_not_block_another(self):
+        for i in range(RATE_LIMIT):
+            self._confirm_login(UserFactory(), ip=f"10.0.1.{i}")
+        response = self._confirm_login(UserFactory(), ip="10.0.1.99")
+        self.assertNotEqual(response.status_code, 403)
+
 
 class TestPasswordResetRateLimit(TestCase):
     """Password reset view is rate-limited by both IP and email address."""

hypha/apply/users/utils.py

@@ -13,6 +13,7 @@
 from django.utils.encoding import force_bytes
 from django.utils.http import url_has_allowed_host_and_scheme, urlsafe_base64_encode
 from django.utils.translation import gettext as _
+from django_ratelimit.core import _get_ip
 
 
 def get_user_by_email(email):
@@ -179,6 +180,22 @@ def generate_numeric_token(length=6):
     return get_random_string(length, allowed_chars=string.digits)
 
 
+def login_ratelimit_key(group, request):
+    """Per-account rate-limit key for the two-factor login wizard.
+
+    The wizard prefixes its fields, so the account identifier arrives as
+    `auth-username`, not `email`. The later steps (OTP, backup token) post no
+    username at all, and neither do the passwordless views that share this
+    decorated `dispatch` — those fall back to the client IP.
+
+    Never return a constant for the missing-field case: django-ratelimit hashes
+    whatever it is given, so every such request would land in a single bucket
+    and any one client could exhaust login for everybody.
+    """
+    username = request.POST.get("auth-username", "").strip().lower()
+    return username or f"ip:{_get_ip(request)}"
+
+
 def update_is_staff(request, user):
     """Determine if the user should have `is_staff`
 

hypha/apply/users/views.py

@@ -72,6 +72,7 @@
     generate_numeric_token,
     get_redirect_url,
     get_zoneinfo,
+    login_ratelimit_key,
     send_activation_email,
     send_confirmation_email,
 )
@@ -84,7 +85,7 @@
     name="dispatch",
 )
 @method_decorator(
-    ratelimit(key="post:email", rate=settings.DEFAULT_RATE_LIMIT, method="POST"),
+    ratelimit(key=login_ratelimit_key, rate=settings.DEFAULT_RATE_LIMIT, method="POST"),
     name="dispatch",
 )
 class LoginView(TwoFactorLoginView):
@@ -178,7 +179,7 @@ def hijack_view(request):
 
 @login_required
 def account_email_change(request):
-    if request.user.has_usable_password() and not request.is_elevated():
+    if not request.is_elevated():
         return redirect_to_elevate(request.get_full_path())
 
     signer = TimestampSigner()