Changeset View
Changeset View
Standalone View
Standalone View
swh/web/api/throttling.py
| # Copyright (C) 2017-2022 The Software Heritage developers | # Copyright (C) 2017-2022 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU Affero General Public License version 3, or any later version | # License: GNU Affero General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from ipaddress import IPv4Network, IPv6Network, ip_address, ip_network | from ipaddress import IPv4Network, IPv6Network, ip_address, ip_network | ||||
| from typing import Callable, List, TypeVar, Union | from typing import Callable, List, TypeVar, Union | ||||
| from django.core.exceptions import ImproperlyConfigured | from django.core.exceptions import ImproperlyConfigured | ||||
| import rest_framework | import rest_framework | ||||
| from rest_framework.throttling import ScopedRateThrottle | from rest_framework.throttling import ScopedRateThrottle | ||||
| from swh.web.auth.utils import API_SAVE_ORIGIN_PERMISSION | from swh.web.auth.utils import API_RAW_OBJECT_PERMISSION, API_SAVE_ORIGIN_PERMISSION | ||||
| from swh.web.common.exc import sentry_capture_exception | from swh.web.common.exc import sentry_capture_exception | ||||
| from swh.web.config import get_config | from swh.web.config import get_config | ||||
| APIView = TypeVar("APIView", bound="rest_framework.views.APIView") | APIView = TypeVar("APIView", bound="rest_framework.views.APIView") | ||||
| Request = rest_framework.request.Request | Request = rest_framework.request.Request | ||||
| API_THROTTLING_EXEMPTED_PERM = "swh.web.api.throttling_exempted" | API_THROTTLING_EXEMPTED_PERM = "swh.web.api.throttling_exempted" | ||||
| ▲ Show 20 Lines • Show All 162 Lines • ▼ Show 20 Lines | def allow_request(self, request: Request, view: APIView) -> bool: | ||||
| # no throttling for staff users or users with adequate permission | # no throttling for staff users or users with adequate permission | ||||
| return True | return True | ||||
| scope = self.get_scope(view) | scope = self.get_scope(view) | ||||
| if scope == "swh_save_origin" and request.user.has_perm( | if scope == "swh_save_origin" and request.user.has_perm( | ||||
| API_SAVE_ORIGIN_PERMISSION | API_SAVE_ORIGIN_PERMISSION | ||||
| ): | ): | ||||
| # no throttling on save origin endpoint for users with adequate permission | # no throttling on save origin endpoint for users with adequate permission | ||||
| return True | return True | ||||
| if scope == "swh_raw_object" and request.user.has_perm( | |||||
| API_RAW_OBJECT_PERMISSION | |||||
| ): | |||||
| # no throttling on raw object endpoint for users with adequate permission | |||||
| return True | |||||
| return super().allow_request(request, view) | return super().allow_request(request, view) | ||||
| def throttle_scope(scope: str) -> Callable[..., APIView]: | def throttle_scope(scope: str) -> Callable[..., APIView]: | ||||
| """Decorator that allows the throttle scope of a DRF | """Decorator that allows the throttle scope of a DRF | ||||
| function based view to be set:: | function based view to be set:: | ||||
| @api_view(['GET', ]) | @api_view(['GET', ]) | ||||
| Show All 19 Lines | |||||