exceptions: Add AccessDeniedError.

zerver/decorator.py

@@ -24,6 +24,7 @@ from django_otp import user_has_device
 from two_factor.utils import default_device
 
 from zerver.lib.exceptions import (
+    AccessDeniedError,
     ErrorCode,
     InvalidAPIKeyError,
     InvalidAPIKeyFormatError,
@@ -39,7 +40,7 @@ from zerver.lib.exceptions import (
 from zerver.lib.queue import queue_json_publish
 from zerver.lib.rate_limiter import RateLimitedUser
 from zerver.lib.request import REQ, has_request_variables
-from zerver.lib.response import json_error, json_method_not_allowed, json_success, json_unauthorized
+from zerver.lib.response import json_method_not_allowed, json_success, json_unauthorized
 from zerver.lib.subdomains import get_subdomain, user_matches_subdomain
 from zerver.lib.timestamp import datetime_to_timestamp, timestamp_to_datetime
 from zerver.lib.types import ViewFuncT
@@ -798,7 +799,7 @@ def internal_notify_view(is_tornado_view: bool) -> Callable[[ViewFuncT], ViewFun
             request: HttpRequest, *args: object, **kwargs: object
         ) -> HttpResponse:
             if not authenticate_notify(request):
-                return json_error(_("Access denied"), status=403)
+                raise AccessDeniedError()
             is_tornado_request = hasattr(request, "_tornado_handler")
             # These next 2 are not security checks; they are internal
             # assertions to help us find bugs.

zerver/lib/exceptions.py

@@ -375,3 +375,14 @@ class InvitationError(JsonableError):
         self.errors: List[Tuple[str, str, bool]] = errors
         self.sent_invitations: bool = sent_invitations
         self.license_limit_reached: bool = license_limit_reached
+
+
+class AccessDeniedError(JsonableError):
+    http_status_code = 403
+
+    def __init__(self) -> None:
+        pass
+
+    @staticmethod
+    def msg_format() -> str:
+        return _("Access denied")

zerver/tests/test_decorators.py

@@ -37,6 +37,7 @@ from zerver.lib.actions import (
 )
 from zerver.lib.cache import dict_to_items_tuple, ignore_unhashable_lru_cache, items_tuple_to_dict
 from zerver.lib.exceptions import (
+    AccessDeniedError,
     InvalidAPIKeyError,
     InvalidAPIKeyFormatError,
     JsonableError,
@@ -1519,7 +1520,9 @@ class TestInternalNotifyView(ZulipTestCase):
 
         with self.settings(SHARED_SECRET="broken"):
             self.assertFalse(authenticate_notify(req))
-            self.assertEqual(self.internal_notify(True, req).status_code, 403)
+            with self.assertRaises(AccessDeniedError) as context:
+                self.internal_notify(True, req)
+            self.assertEqual(context.exception.http_status_code, 403)
 
     def test_external_requests(self) -> None:
         secret = "random"
@@ -1530,7 +1533,9 @@ class TestInternalNotifyView(ZulipTestCase):
 
         with self.settings(SHARED_SECRET=secret):
             self.assertFalse(authenticate_notify(req))
-            self.assertEqual(self.internal_notify(True, req).status_code, 403)
+            with self.assertRaises(AccessDeniedError) as context:
+                self.internal_notify(True, req)
+            self.assertEqual(context.exception.http_status_code, 403)
 
     def test_is_local_address(self) -> None:
         self.assertTrue(is_local_addr("127.0.0.1"))

zerver/tests/test_event_system.py

@@ -16,6 +16,7 @@ from zerver.lib.actions import (
 )
 from zerver.lib.event_schema import check_restart_event
 from zerver.lib.events import fetch_initial_state_data, get_raw_user_data
+from zerver.lib.exceptions import AccessDeniedError
 from zerver.lib.test_classes import ZulipTestCase
 from zerver.lib.test_helpers import HostRequestMock, queries_captured, stub_event_queue_user_events
 from zerver.lib.users import get_api_key
@@ -190,8 +191,10 @@ class EventsEndpointTest(ZulipTestCase):
         )
         req = HostRequestMock(post_data, user_profile=None)
         req.META["REMOTE_ADDR"] = "127.0.0.1"
+        with self.assertRaises(AccessDeniedError) as context:
             result = self.client_post_request("/notify_tornado", req)
-        self.assert_json_error(result, "Access denied", status_code=403)
+        self.assertEqual(str(context.exception), "Access denied")
+        self.assertEqual(context.exception.http_status_code, 403)
 
         post_data["secret"] = settings.SHARED_SECRET
         req = HostRequestMock(post_data, user_profile=None)