Add high, medium, and low tier based throttling permissions

Signed-off-by: Keshav Priyadarshi <git@keshav.space>

Author: keshav-space

vulnerabilities/api.py

@@ -22,7 +22,6 @@
 from rest_framework import viewsets
 from rest_framework.decorators import action
 from rest_framework.response import Response
-from rest_framework.throttling import AnonRateThrottle
 
 from vulnerabilities.models import Alias
 from vulnerabilities.models import Exploit
@@ -471,7 +470,7 @@ class PackageViewSet(viewsets.ReadOnlyModelViewSet):
     serializer_class = PackageSerializer
     filter_backends = (filters.DjangoFilterBackend,)
     filterset_class = PackageFilterSet
-    throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
+    throttle_classes = [PermissionBasedUserRateThrottle]
 
     def get_queryset(self):
         return super().get_queryset().with_is_vulnerable()
@@ -688,7 +687,7 @@ def get_queryset(self):
     serializer_class = VulnerabilitySerializer
     filter_backends = (filters.DjangoFilterBackend,)
     filterset_class = VulnerabilityFilterSet
-    throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
+    throttle_classes = [PermissionBasedUserRateThrottle]
 
 
 class CPEFilterSet(filters.FilterSet):

vulnerabilities/api_extension.py

@@ -23,7 +23,6 @@
 from rest_framework.serializers import ModelSerializer
 from rest_framework.serializers import Serializer
 from rest_framework.serializers import ValidationError
-from rest_framework.throttling import AnonRateThrottle
 
 from vulnerabilities.api import BaseResourceSerializer
 from vulnerabilities.models import Exploit
@@ -259,7 +258,7 @@ class V2PackageViewSet(viewsets.ReadOnlyModelViewSet):
     lookup_field = "purl"
     filter_backends = (filters.DjangoFilterBackend,)
     filterset_class = V2PackageFilterSet
-    throttle_classes = [PermissionBasedUserRateThrottle, AnonRateThrottle]
+    throttle_classes = [PermissionBasedUserRateThrottle]
 
     def get_queryset(self):
         return super().get_queryset().with_is_vulnerable().prefetch_related("vulnerabilities")
@@ -345,7 +344,7 @@ class VulnerabilityViewSet(viewsets.ReadOnlyModelViewSet):
     lookup_field = "vulnerability_id"
     filter_backends = (filters.DjangoFilterBackend,)
     filterset_class = V2VulnerabilityFilterSet
-    throttle_classes = [PermissionBasedUserRateThrottle, AnonRateThrottle]
+    throttle_classes = [PermissionBasedUserRateThrottle]
 
     def get_queryset(self):
         """
@@ -381,7 +380,7 @@ class CPEViewSet(viewsets.ReadOnlyModelViewSet):
     ).distinct()
     serializer_class = V2VulnerabilitySerializer
     filter_backends = (filters.DjangoFilterBackend,)
-    throttle_classes = [PermissionBasedUserRateThrottle, AnonRateThrottle]
+    throttle_classes = [PermissionBasedUserRateThrottle]
     filterset_class = CPEFilterSet
 
     @action(detail=False, methods=["post"])
@@ -420,4 +419,4 @@ class AliasViewSet(viewsets.ReadOnlyModelViewSet):
     serializer_class = V2VulnerabilitySerializer
     filter_backends = (filters.DjangoFilterBackend,)
     filterset_class = AliasFilterSet
-    throttle_classes = [PermissionBasedUserRateThrottle, AnonRateThrottle]
+    throttle_classes = [PermissionBasedUserRateThrottle]

vulnerabilities/migrations/0093_alter_apiuser_options.py

@@ -1,4 +1,4 @@
-# Generated by Django 4.2.22 on 2025-06-13 12:44
+# Generated by Django 4.2.22 on 2025-06-25 18:56
 
 from django.db import migrations
 
@@ -14,10 +14,22 @@ class Migration(migrations.Migration):
             name="apiuser",
             options={
                 "permissions": [
-                    ("throttle_unrestricted", "Can make api requests without throttling limits"),
-                    ("throttle_18000_hour", "Can make 18000 api requests per hour"),
-                    ("throttle_14400_hour", "Can make 14400 api requests per hour"),
-                    ("throttle_3600_hour", "Can make 3600 api requests per hour"),
+                    (
+                        "throttle_3_unrestricted",
+                        "Can make unlimited API requests without any throttling limits",
+                    ),
+                    (
+                        "throttle_2_high",
+                        "Can make high number of API requests with minimal throttling",
+                    ),
+                    (
+                        "throttle_1_medium",
+                        "Can make medium number of API requests with standard throttling",
+                    ),
+                    (
+                        "throttle_0_low",
+                        "Can make low number of API requests with strict throttling",
+                    ),
                 ]
             },
         ),

vulnerabilities/models.py

@@ -1497,10 +1497,22 @@ class ApiUser(UserModel):
     class Meta:
         proxy = True
         permissions = [
-            ("throttle_unrestricted", "Can make api requests without throttling limits"),
-            ("throttle_18000_hour", "Can make 18000 api requests per hour"),
-            ("throttle_14400_hour", "Can make 14400 api requests per hour"),
-            ("throttle_3600_hour", "Can make 3600 api requests per hour"),
+            (
+                "throttle_3_unrestricted",
+                "Can make unlimited API requests without any throttling limits",
+            ),
+            (
+                "throttle_2_high",
+                "Can make high number of API requests with minimal throttling",
+            ),
+            (
+                "throttle_1_medium",
+                "Can make medium number of API requests with standard throttling",
+            ),
+            (
+                "throttle_0_low",
+                "Can make low number of API requests with strict throttling",
+            ),
         ]
 
 

vulnerabilities/tests/test_throttling.py

@@ -14,22 +14,17 @@
 from rest_framework import status
 from rest_framework.test import APIClient
 from rest_framework.test import APITestCase
-from rest_framework.throttling import AnonRateThrottle
 
 from vulnerabilities.api import PermissionBasedUserRateThrottle
 from vulnerabilities.models import ApiUser
 
 
-def simulate_throttle_usage(
-    url,
-    client,
-    mock_use_count,
-    throttle_cls=PermissionBasedUserRateThrottle,
-):
-    throttle = throttle_cls()
+def simulate_throttle_usage(url, client, mock_use_count):
+    throttle = PermissionBasedUserRateThrottle()
     request = client.get(url).wsgi_request
 
     if cache_key := throttle.get_cache_key(request, view=None):
+        print(cache_key)
         now = throttle.timer()
         cache.set(cache_key, [now] * mock_use_count)
 
@@ -41,37 +36,37 @@ def setUp(self):
         # See https://www.django-rest-framework.org/api-guide/throttling/#setting-up-the-cache
         cache.clear()
 
-        permission_3600 = Permission.objects.get(codename="throttle_3600_hour")
-        permission_14400 = Permission.objects.get(codename="throttle_14400_hour")
-        permission_18000 = Permission.objects.get(codename="throttle_18000_hour")
-        permission_unrestricted = Permission.objects.get(codename="throttle_unrestricted")
+        permission_low = Permission.objects.get(codename="throttle_0_low")
+        permission_medium = Permission.objects.get(codename="throttle_1_medium")
+        permission_high = Permission.objects.get(codename="throttle_2_high")
+        permission_unrestricted = Permission.objects.get(codename="throttle_3_unrestricted")
 
-        # user with 3600/hour permission
-        self.th_3600_user = ApiUser.objects.create_api_user(username="z@mail.com")
-        self.th_3600_user.user_permissions.add(permission_3600)
-        self.th_3600_user_auth = f"Token {self.th_3600_user.auth_token.key}"
-        self.th_3600_user_csrf_client = APIClient(enforce_csrf_checks=True)
-        self.th_3600_user_csrf_client.credentials(HTTP_AUTHORIZATION=self.th_3600_user_auth)
+        # user with low permission
+        self.th_low_user = ApiUser.objects.create_api_user(username="z@mail.com")
+        self.th_low_user.user_permissions.add(permission_low)
+        self.th_low_user_auth = f"Token {self.th_low_user.auth_token.key}"
+        self.th_low_user_csrf_client = APIClient(enforce_csrf_checks=True)
+        self.th_low_user_csrf_client.credentials(HTTP_AUTHORIZATION=self.th_low_user_auth)
 
         # basic user without any special throttling perm
         self.basic_user = ApiUser.objects.create_api_user(username="a@mail.com")
         self.basic_user_auth = f"Token {self.basic_user.auth_token.key}"
         self.basic_user_csrf_client = APIClient(enforce_csrf_checks=True)
         self.basic_user_csrf_client.credentials(HTTP_AUTHORIZATION=self.basic_user_auth)
 
-        # 14400/hour permission
-        self.th_14400_user = ApiUser.objects.create_api_user(username="b@mail.com")
-        self.th_14400_user.user_permissions.add(permission_14400)
-        self.th_14400_user_auth = f"Token {self.th_14400_user.auth_token.key}"
-        self.th_14400_user_csrf_client = APIClient(enforce_csrf_checks=True)
-        self.th_14400_user_csrf_client.credentials(HTTP_AUTHORIZATION=self.th_14400_user_auth)
+        # medium permission
+        self.th_medium_user = ApiUser.objects.create_api_user(username="b@mail.com")
+        self.th_medium_user.user_permissions.add(permission_medium)
+        self.th_medium_user_auth = f"Token {self.th_medium_user.auth_token.key}"
+        self.th_medium_user_csrf_client = APIClient(enforce_csrf_checks=True)
+        self.th_medium_user_csrf_client.credentials(HTTP_AUTHORIZATION=self.th_medium_user_auth)
 
-        # 18000/hour permission
-        self.th_18000_user = ApiUser.objects.create_api_user(username="c@mail.com")
-        self.th_18000_user.user_permissions.add(permission_18000)
-        self.th_18000_user_auth = f"Token {self.th_18000_user.auth_token.key}"
-        self.th_18000_user_csrf_client = APIClient(enforce_csrf_checks=True)
-        self.th_18000_user_csrf_client.credentials(HTTP_AUTHORIZATION=self.th_18000_user_auth)
+        # high permission
+        self.th_high_user = ApiUser.objects.create_api_user(username="c@mail.com")
+        self.th_high_user.user_permissions.add(permission_high)
+        self.th_high_user_auth = f"Token {self.th_high_user.auth_token.key}"
+        self.th_high_user_csrf_client = APIClient(enforce_csrf_checks=True)
+        self.th_high_user_csrf_client.credentials(HTTP_AUTHORIZATION=self.th_high_user_auth)
 
         # unrestricted throttling perm
         self.th_unrestricted_user = ApiUser.objects.create_api_user(username="d@mail.com")
@@ -85,60 +80,60 @@ def setUp(self):
         self.csrf_client_anon = APIClient(enforce_csrf_checks=True)
         self.csrf_client_anon_1 = APIClient(enforce_csrf_checks=True)
 
-    def test_user_with_3600_perm_throttling(self):
+    def test_user_with_low_perm_throttling(self):
         simulate_throttle_usage(
             url="/api/packages",
-            client=self.th_3600_user_csrf_client,
-            mock_use_count=3599,
+            client=self.th_low_user_csrf_client,
+            mock_use_count=10799,
         )
 
-        response = self.th_3600_user_csrf_client.get("/api/packages")
+        response = self.th_low_user_csrf_client.get("/api/packages")
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
-        # exhausted 3600/hr allowed requests.
-        response = self.th_3600_user_csrf_client.get("/api/packages")
+        # exhausted 10800/hr allowed requests.
+        response = self.th_low_user_csrf_client.get("/api/packages")
         self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
 
     def test_basic_user_throttling(self):
         simulate_throttle_usage(
             url="/api/packages",
             client=self.basic_user_csrf_client,
-            mock_use_count=10799,
+            mock_use_count=14399,
         )
 
         response = self.basic_user_csrf_client.get("/api/packages")
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
-        # exhausted 10800/hr allowed requests.
+        # exhausted 14400/hr allowed requests.
         response = self.basic_user_csrf_client.get("/api/packages")
         self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
 
-    def test_user_with_14400_perm_throttling(self):
+    def test_user_with_medium_perm_throttling(self):
         simulate_throttle_usage(
             url="/api/packages",
-            client=self.th_14400_user_csrf_client,
+            client=self.th_medium_user_csrf_client,
             mock_use_count=14399,
         )
 
-        response = self.th_14400_user_csrf_client.get("/api/packages")
+        response = self.th_medium_user_csrf_client.get("/api/packages")
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
         # exhausted 14400/hr allowed requests for user with 14400 perm.
-        response = self.th_14400_user_csrf_client.get("/api/packages")
+        response = self.th_medium_user_csrf_client.get("/api/packages")
         self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
 
-    def test_user_with_18000_perm_throttling(self):
+    def test_user_with_high_perm_throttling(self):
         simulate_throttle_usage(
             url="/api/packages",
-            client=self.th_18000_user_csrf_client,
+            client=self.th_high_user_csrf_client,
             mock_use_count=17999,
         )
 
-        response = self.th_18000_user_csrf_client.get("/api/packages")
+        response = self.th_high_user_csrf_client.get("/api/packages")
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
         # exhausted 18000/hr allowed requests for user with 18000 perm.
-        response = self.th_18000_user_csrf_client.get("/api/packages")
+        response = self.th_high_user_csrf_client.get("/api/packages")
         self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
 
     def test_user_with_unrestricted_perm_throttling(self):
@@ -154,7 +149,6 @@ def test_user_with_unrestricted_perm_throttling(self):
 
     def test_anon_throttling(self):
         simulate_throttle_usage(
-            throttle_cls=AnonRateThrottle,
             url="/api/packages",
             client=self.csrf_client_anon,
             mock_use_count=3599,

vulnerabilities/throttling.py

@@ -7,35 +7,49 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
+from django.core.exceptions import ImproperlyConfigured
 from rest_framework.exceptions import Throttled
 from rest_framework.throttling import UserRateThrottle
 from rest_framework.views import exception_handler
 
 
 class PermissionBasedUserRateThrottle(UserRateThrottle):
     """
-    Throttle authenticated users based on their assigned permissions.
-    If no throttling permission is assigned, default to rate for `user`
-    scope provided via `DEFAULT_THROTTLE_RATES` in settings.py.
+    Throttles authenticated users based on their assigned permissions.
+    If no throttling permission is assigned, defaults to `medium` throttling
+    for authenticated users and `anon` for unauthenticated users.
     """
 
+    def __init__(self):
+        pass
+
     def allow_request(self, request, view):
         user = request.user
+        throttling_tier = "medium"
 
-        if user and user.is_authenticated:
-            if user.has_perm("vulnerabilities.throttle_unrestricted"):
-                return True
-            elif user.has_perm("vulnerabilities.throttle_18000_hour"):
-                self.rate = "18000/hour"
-            elif user.has_perm("vulnerabilities.throttle_14400_hour"):
-                self.rate = "14400/hour"
-            elif user.has_perm("vulnerabilities.throttle_3600_hour"):
-                self.rate = "3600/hour"
+        if not user or not user.is_authenticated:
+            throttling_tier = "anon"
+        elif user.has_perm("vulnerabilities.throttle_3_unrestricted"):
+            return True
+        elif user.has_perm("vulnerabilities.throttle_2_high"):
+            throttling_tier = "high"
+        elif user.has_perm("vulnerabilities.throttle_1_medium"):
+            throttling_tier = "medium"
+        elif user.has_perm("vulnerabilities.throttle_0_low"):
+            throttling_tier = "low"
 
-            self.num_requests, self.duration = self.parse_rate(self.rate)
+        self.rate = self.get_throttle_rate(throttling_tier)
+        self.num_requests, self.duration = self.parse_rate(self.rate)
 
         return super().allow_request(request, view)
 
+    def get_throttle_rate(self, tier):
+        try:
+            return self.THROTTLE_RATES[tier]
+        except KeyError:
+            msg = f"No throttle rate set for {tier}."
+            raise ImproperlyConfigured(msg)
+
 
 def throttled_exception_handler(exception, context):
     """

vulnerablecode/settings.py

@@ -190,7 +190,17 @@
 LOGIN_REDIRECT_URL = "/"
 LOGOUT_REDIRECT_URL = "/"
 
-REST_FRAMEWORK_DEFAULT_THROTTLE_RATES = {"anon": "3600/hour", "user": "10800/hour"}
+THROTTLE_RATES_ANON = env.str("THROTTLE_RATES_ANON", default="3600/hour")
+THROTTLE_RATES_USER_HIGH = env.str("THROTTLE_RATES_USER_HIGH", default="18000/hour")
+THROTTLE_RATES_USER_MEDIUM = env.str("THROTTLE_RATES_USER_MEDIUM", default="14400/hour")
+THROTTLE_RATES_USER_LOW = env.str("THROTTLE_RATES_USER_LOW", default="10800/hour")
+
+REST_FRAMEWORK_DEFAULT_THROTTLE_RATES = {
+    "anon": THROTTLE_RATES_ANON,
+    "low": THROTTLE_RATES_USER_LOW,
+    "medium": THROTTLE_RATES_USER_MEDIUM,
+    "high": THROTTLE_RATES_USER_HIGH,
+}
 
 
 if IS_TESTS:
@@ -235,7 +245,6 @@
     ),
     "DEFAULT_THROTTLE_CLASSES": [
         "vulnerabilities.throttling.PermissionBasedUserRateThrottle",
-        "rest_framework.throttling.AnonRateThrottle",
     ],
     "DEFAULT_THROTTLE_RATES": REST_FRAMEWORK_DEFAULT_THROTTLE_RATES,
     "EXCEPTION_HANDLER": "vulnerabilities.throttling.throttled_exception_handler",