diff --git a/openedx/core/djangoapps/notifications/email/utils.py b/openedx/core/djangoapps/notifications/email/utils.py index 3ce2590d18..1f761c0860 100644 --- a/openedx/core/djangoapps/notifications/email/utils.py +++ b/openedx/core/djangoapps/notifications/email/utils.py @@ -6,13 +6,14 @@ import datetime from bs4 import BeautifulSoup from django.conf import settings from django.contrib.auth import get_user_model +from django.core.exceptions import BadRequest from django.shortcuts import get_object_or_404 from django.utils.translation import gettext as _ from pytz import utc from waffle import get_waffle_flag_model # pylint: disable=invalid-django-waffle-import from lms.djangoapps.branding.api import get_logo_url_for_email -from lms.djangoapps.discussion.notification_prefs.views import UsernameCipher +from lms.djangoapps.discussion.notification_prefs.views import UsernameCipher, UsernameDecryptionException from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY from openedx.core.djangoapps.notifications.base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES from openedx.core.djangoapps.notifications.config.waffle import ENABLE_EMAIL_NOTIFICATIONS @@ -384,6 +385,19 @@ def decrypt_string(string): return UsernameCipher.decrypt(string).decode() +def username_from_hash(group, request): + """ + Django ratelimit key to return username from hash + """ + username = request.resolver_match.kwargs.get("username") + if username: + try: + return decrypt_string(username) + except UsernameDecryptionException as exc: + raise BadRequest("Bad request") from exc + return None + + def update_user_preferences_from_patch(encrypted_username): """ Decrypt username and patch and updates user preferences diff --git a/openedx/core/djangoapps/notifications/tests/test_views.py b/openedx/core/djangoapps/notifications/tests/test_views.py index de02141745..d5b2237303 100644 --- a/openedx/core/djangoapps/notifications/tests/test_views.py +++ b/openedx/core/djangoapps/notifications/tests/test_views.py @@ -7,6 +7,7 @@ from unittest import mock import ddt from django.conf import settings from django.contrib.auth import get_user_model +from django.core.cache import cache from django.test.utils import override_settings from django.urls import reverse from edx_toggles.toggles.testutils import override_waffle_flag @@ -481,6 +482,7 @@ class UpdatePreferenceFromEncryptedDataView(ModuleStoreTestCase): """ Setup test case """ + cache.clear() super().setUp() password = 'password' self.user = UserFactory(password=password) @@ -488,6 +490,22 @@ class UpdatePreferenceFromEncryptedDataView(ModuleStoreTestCase): self.course = CourseFactory.create(display_name='test course 1', run="Testing_course_1") CourseNotificationPreference(course_id=self.course.id, user=self.user).save() + @override_settings(LMS_BASE="example.com", ONE_CLICK_UNSUBSCRIBE_RATE_LIMIT='1/d') + def test_rate_limit_on_unsub(self): + """ + Test rate limit on unsub + """ + self.client.logout() + user_hash = encrypt_string(self.user.username) + url_params = { + "username": user_hash, + } + url = reverse("preference_update_view", kwargs=url_params) + response = self.client.get(url) + assert response.status_code == status.HTTP_200_OK + response = self.client.get(url) + assert response.status_code == status.HTTP_429_TOO_MANY_REQUESTS + @override_settings(LMS_BASE="") @ddt.data('get', 'post') def test_if_preference_is_updated(self, request_type): diff --git a/openedx/core/djangoapps/notifications/views.py b/openedx/core/djangoapps/notifications/views.py index d3a9dd1f48..57ef88cde1 100644 --- a/openedx/core/djangoapps/notifications/views.py +++ b/openedx/core/djangoapps/notifications/views.py @@ -5,6 +5,7 @@ from datetime import datetime, timedelta from django.conf import settings from django.db.models import Count +from django_ratelimit.core import is_ratelimited from django.shortcuts import get_object_or_404 from django.utils.translation import gettext as _ from pytz import UTC @@ -14,7 +15,7 @@ from rest_framework.generics import UpdateAPIView from rest_framework.response import Response from rest_framework.views import APIView -from openedx.core.djangoapps.notifications.email.utils import update_user_preferences_from_patch +from openedx.core.djangoapps.notifications.email.utils import update_user_preferences_from_patch, username_from_hash from openedx.core.djangoapps.notifications.models import NotificationPreference from openedx.core.djangoapps.notifications.permissions import allow_any_authenticated_user @@ -241,6 +242,11 @@ def preference_update_from_encrypted_username_view(request, username, patch=""): View to update user preferences from encrypted username and patch. username and patch must be string """ + if is_ratelimited( + request=request, group="unsubscribe", key=username_from_hash, + rate=settings.ONE_CLICK_UNSUBSCRIBE_RATE_LIMIT, increment=True, + ): + return Response({"error": "Too many requests"}, status=status.HTTP_429_TOO_MANY_REQUESTS) update_user_preferences_from_patch(username) return Response({"result": "success"}, status=status.HTTP_200_OK) diff --git a/openedx/envs/common.py b/openedx/envs/common.py index 767bfdbd3e..c888ec8fd7 100644 --- a/openedx/envs/common.py +++ b/openedx/envs/common.py @@ -828,6 +828,8 @@ USERNAME_PATTERN = fr'(?P{USERNAME_REGEX_PARTIAL})' DISCUSSION_RATELIMIT = '100/m' SKIP_RATE_LIMIT_ON_ACCOUNT_AFTER_DAYS = 0 +ONE_CLICK_UNSUBSCRIBE_RATE_LIMIT = '100/m' + LMS_ROOT_URL = None LMS_INTERNAL_ROOT_URL = Derived(lambda settings: settings.LMS_ROOT_URL)