add APIGW NG workaround for AccessDenied exception (#11317)

bentsku · web-flow · commit 7f60a433ff73 · 2024-08-06T18:18:54.000+02:00
diff --git a/localstack-core/localstack/services/apigateway/next_gen/execute_api/handlers/gateway_exception.py b/localstack-core/localstack/services/apigateway/next_gen/execute_api/handlers/gateway_exception.py
@@ -11,6 +11,7 @@
 )
 from localstack.services.apigateway.next_gen.execute_api.context import RestApiInvocationContext
 from localstack.services.apigateway.next_gen.execute_api.gateway_response import (
+    AccessDeniedError,
     BaseGatewayException,
     get_gateway_response_or_default,
 )
@@ -81,6 +82,13 @@ def create_exception_response(
     def _build_response_content(exception: BaseGatewayException) -> str:
         # TODO apply responseTemplates to the content. We should also handle the default simply by managing the default
         #  template body `{"message":$context.error.messageString}`
+
+        # TODO: remove this workaround by properly managing the responseTemplate for UnauthorizedError
+        #  on the CRUD level, it returns the same template as all other errors but in reality the message field is
+        #  capitalized
+        if isinstance(exception, AccessDeniedError):
+            return json.dumps({"Message": exception.message})
+
         return json.dumps({"message": exception.message})
 
     @staticmethod
diff --git a/tests/unit/services/apigateway/test_handler_exception.py b/tests/unit/services/apigateway/test_handler_exception.py
@@ -8,6 +8,7 @@
 from localstack.services.apigateway.next_gen.execute_api.gateway_response import (
     AccessDeniedError,
     BaseGatewayException,
+    UnauthorizedError,
 )
 from localstack.services.apigateway.next_gen.execute_api.handlers import GatewayExceptionHandler
 from localstack.services.apigateway.next_gen.execute_api.router import ApiGatewayEndpoint
@@ -69,27 +70,27 @@ def test_non_gateway_exception(self, get_context, apigw_response):
     def test_gateway_exception(self, get_context, apigw_response):
         exception_handler = GatewayExceptionHandler()
 
-        # Create an Access Denied exception with no Gateway Response configured
-        exception = AccessDeniedError("Access Denied")
+        # Create an UnauthorizedError exception with no Gateway Response configured
+        exception = UnauthorizedError("Unauthorized")
         exception_handler(
             chain=RestApiGatewayHandlerChain(),
             exception=exception,
             context=get_context(),
             response=apigw_response,
         )
 
-        assert apigw_response.status_code == 403
-        assert apigw_response.json == {"message": "Access Denied"}
-        assert apigw_response.headers.get("x-amzn-errortype") == "AccessDeniedException"
+        assert apigw_response.status_code == 401
+        assert apigw_response.json == {"message": "Unauthorized"}
+        assert apigw_response.headers.get("x-amzn-errortype") == "UnauthorizedException"
 
     def test_gateway_exception_with_default_4xx(self, get_context, apigw_response):
         exception_handler = GatewayExceptionHandler()
 
         # Configure DEFAULT_4XX response
         gateway_responses = {GatewayResponseType.DEFAULT_4XX: GatewayResponse(statusCode="400")}
 
-        # Create an Access Denied exception with DEFAULT_4xx configured
-        exception = AccessDeniedError("Access Denied")
+        # Create an UnauthorizedError exception with DEFAULT_4xx configured
+        exception = UnauthorizedError("Unauthorized")
         exception_handler(
             chain=RestApiGatewayHandlerChain(),
             exception=exception,
@@ -98,24 +99,41 @@ def test_gateway_exception_with_default_4xx(self, get_context, apigw_response):
         )
 
         assert apigw_response.status_code == 400
-        assert apigw_response.json == {"message": "Access Denied"}
-        assert apigw_response.headers.get("x-amzn-errortype") == "AccessDeniedException"
+        assert apigw_response.json == {"message": "Unauthorized"}
+        assert apigw_response.headers.get("x-amzn-errortype") == "UnauthorizedException"
 
     def test_gateway_exception_with_gateway_response(self, get_context, apigw_response):
         exception_handler = GatewayExceptionHandler()
 
         # Configure Access Denied response
-        gateway_responses = {GatewayResponseType.ACCESS_DENIED: GatewayResponse(statusCode="400")}
+        gateway_responses = {GatewayResponseType.UNAUTHORIZED: GatewayResponse(statusCode="405")}
 
-        # Create an Access Denied exception with ACCESS_DENIED configured
-        exception = AccessDeniedError("Access Denied")
+        # Create an UnauthorizedError exception with UNAUTHORIZED configured
+        exception = UnauthorizedError("Unauthorized")
         exception_handler(
             chain=RestApiGatewayHandlerChain(),
             exception=exception,
             context=get_context(gateway_responses),
             response=apigw_response,
         )
 
-        assert apigw_response.status_code == 400
-        assert apigw_response.json == {"message": "Access Denied"}
+        assert apigw_response.status_code == 405
+        assert apigw_response.json == {"message": "Unauthorized"}
+        assert apigw_response.headers.get("x-amzn-errortype") == "UnauthorizedException"
+
+    def test_gateway_exception_access_denied(self, get_context, apigw_response):
+        # special case where the `Message` field is capitalized
+        exception_handler = GatewayExceptionHandler()
+
+        # Create an AccessDeniedError exception with no Gateway Response configured
+        exception = AccessDeniedError("Access Denied")
+        exception_handler(
+            chain=RestApiGatewayHandlerChain(),
+            exception=exception,
+            context=get_context(),
+            response=apigw_response,
+        )
+
+        assert apigw_response.status_code == 403
+        assert apigw_response.json == {"Message": "Access Denied"}
         assert apigw_response.headers.get("x-amzn-errortype") == "AccessDeniedException"
