import ipaddress from rest_framework.views import APIView from rest_framework.permissions import IsAdminUser from rest_framework.response import Response from django.conf import settings from .renderers import AdminJSONRenderer from pingcap_sso.utils.requests import get_client_ip, is_vpn_ip class Endpoint(APIView): def convert_args(self, request, *args, **kwargs): return (args, kwargs) def dispatch(self, request, *args, **kwargs): self.args = args self.kwargs = kwargs request = self.initialize_request(request, *args, **kwargs) self.request = request self.headers = self.default_response_headers # deprecate? try: self.initial(request, *args, **kwargs) # Get the appropriate handler method if request.method.lower() in self.http_method_names: handler = getattr(self, request.method.lower(), self.http_method_not_allowed) (args, kwargs) = self.convert_args(request, *args, **kwargs) self.args = args self.kwargs = kwargs else: handler = self.http_method_not_allowed response = handler(request, *args, **kwargs) except Exception as exc: response = self.handle_exception(exc) self.response = self.finalize_response( request, response, *args, **kwargs) return self.response class AdminEndpoint(Endpoint): renderer_classes = (AdminJSONRenderer, ) permission_classes = (IsAdminUser, ) def dispatch(self, request, *args, **kwargs): self.args = args self.kwargs = kwargs request = self.initialize_request(request, *args, **kwargs) self.request = request self.headers = self.default_response_headers # deprecate? try: self.initial(request, *args, **kwargs) # Get the appropriate handler method if request.method.lower() in self.http_method_names: handler = getattr(self, request.method.lower(), self.http_method_not_allowed) (args, kwargs) = self.convert_args(request, *args, **kwargs) self.args = args self.kwargs = kwargs else: handler = self.http_method_not_allowed if not is_vpn_ip(get_client_ip(request)): response = Response({ 'error': "Please use VPN to access this API", }, status=403) else: response = handler(request, *args, **kwargs) except Exception as exc: response = self.handle_exception(exc) self.response = self.finalize_response( request, response, *args, **kwargs) return self.response class AskTUGAdminEndpoint(Endpoint): renderer_classes = (AdminJSONRenderer, ) permission_classes = (IsAdminUser, )