diff --git a/hypha/apply/users/tests/test_email_change.py b/hypha/apply/users/tests/test_email_change.py index 863e7fc4a6..35c02c044d 100644 --- a/hypha/apply/users/tests/test_email_change.py +++ b/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") diff --git a/hypha/apply/users/tests/test_ratelimit.py b/hypha/apply/users/tests/test_ratelimit.py index f689af36b7..c6822b6d5e 100644 --- a/hypha/apply/users/tests/test_ratelimit.py +++ b/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,9 +14,14 @@ 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={ @@ -21,6 +29,7 @@ def _post_login(self, email="test@example.com"): "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.""" diff --git a/hypha/apply/users/utils.py b/hypha/apply/users/utils.py index a3271a6474..818cd2ddd3 100644 --- a/hypha/apply/users/utils.py +++ b/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` diff --git a/hypha/apply/users/views.py b/hypha/apply/users/views.py index e10aac90ac..6f1d6b1b60 100644 --- a/hypha/apply/users/views.py +++ b/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()