feat: added rate limit on one click unsubscribe api

openedx/core/djangoapps/notifications/email/utils.py

@@ -6,13 +6,14 @@
 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

openedx/core/djangoapps/notifications/tests/test_views.py

@@ -7,6 +7,7 @@
 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,13 +482,30 @@ def setUp(self):
         """
         Setup test case
         """
+        cache.clear()
         super().setUp()
         password = 'password'
         self.user = UserFactory(password=password)
         self.client.login(username=self.user.username, password=password)
         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):

openedx/core/djangoapps/notifications/views.py

@@ -5,6 +5,7 @@
 
 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.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)
 

openedx/envs/common.py

@@ -828,6 +828,8 @@ def _make_locale_paths(settings):
 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)