localstack · viren-nadkarni · Dec 1, 2025 · Nov 25, 2025 · Nov 25, 2025 · Nov 28, 2025
@@ -1,7 +1,6 @@
 """Handlers for logging."""
 
 import logging
-import types
 from functools import cached_property
 
 from localstack.aws.api import RequestContext, ServiceException
@@ -25,10 +24,14 @@ def __init__(self, logger=None):
         try:
             import moto.core.exceptions
 
-            self._moto_service_exception = moto.core.exceptions.ServiceException
+            self._skip_exceptions(
+                ServiceException,
+                moto.core.exceptions.ServiceException,
+                moto.core.exceptions.RESTError,
+            )
         except (ModuleNotFoundError, AttributeError):
             # Moto may not be available in stripped-down versions of LocalStack, like LocalStack S3 image.
-            self._moto_service_exception = types.EllipsisType
+            self._skip_exceptions = (ServiceException,)
 
     def __call__(
         self,
@@ -37,11 +40,12 @@ def __call__(
         context: RequestContext,
         response: Response,
     ):
-        if isinstance(exception, (ServiceException, self._moto_service_exception)):
+        if isinstance(exception, self._skip_exceptions):
             # We do not want to log an error/stacktrace if the handler is working as expected, but chooses to throw
             # a service exception. It may also throw a Moto ServiceException, which should not be logged either
             # because ServiceExceptionHandler understands it.
             return
+
         if self.logger.isEnabledFor(level=logging.DEBUG):
             self.logger.exception("exception during call chain", exc_info=exception)
         else:

@@ -158,14 +158,17 @@ class ServiceExceptionSerializer(ExceptionHandler):
     def __init__(self):
         self.handle_internal_failures = True
 
+        self._moto_service_exception = types.EllipsisType
+        self._moto_rest_error = types.EllipsisType
+
         try:
             import moto.core.exceptions
 
             self._moto_service_exception = moto.core.exceptions.ServiceException
+            self._moto_rest_error = moto.core.exceptions.RESTError
         except (ModuleNotFoundError, AttributeError) as exc:
             # Moto may not be available in stripped-down versions of LocalStack, like LocalStack S3 image.
-            LOG.debug("Unable to set up Moto ServiceException translation: %s", exc)
-            self._moto_service_exception = types.EllipsisType
+            LOG.debug("Unable to set up Moto exception translation: %s", exc)
 
     def __call__(
         self,
@@ -200,6 +203,12 @@ def create_exception_response(self, exception: Exception, context: RequestContex
                 code=exception.code,
                 message=exception.message,
             )
+        elif isinstance(exception, self._moto_rest_error):
+            # Some Moto exceptions (like ones raised by EC2) are of type RESTError.
+            error = CommonServiceException(
+                code=exception.error_type,
+                message=exception.message,
+            )
 
         elif not isinstance(exception, ServiceException):
             if not self.handle_internal_failures:

@@ -1,5 +1,8 @@
+from collections.abc import Iterable
+
 import pytest
-from moto.ses.exceptions import MessageRejectedError
+from moto.core.exceptions import RESTError, ServiceException
+from moto.ec2.exceptions import EC2_ERROR_RESPONSE
 
 from localstack.aws.api import CommonServiceException, RequestContext
 from localstack.aws.chain import HandlerChain
@@ -200,7 +203,12 @@ def capture_original_exception_handler(
         assert err_context.service_exception.__traceback__ == raised_exception.__traceback__
         assert err_context.service_exception.status_code == 500
 
-    def test_moto_exception_is_parsed(self, service_response_handler_chain):
+    def test_moto_service_exception_is_translated(self, service_response_handler_chain):
+        # Redefine exception here but use the right base exc. This is to improve tolerance against Moto refactors.
+        class MessageRejectedError(ServiceException):
+            code = "MessageRejected"
+
+        # Ensure ServiceExceptions are translated
         context = create_aws_request_context(
             "ses",
             "SendRawEmail",
@@ -221,3 +229,42 @@ def test_moto_exception_is_parsed(self, service_response_handler_chain):
         assert context.service_exception.code == "MessageRejected"
         assert not context.service_exception.sender_fault
         assert context.service_exception.status_code == 400
+
+    def test_moto_rest_error_is_translated(self, service_response_handler_chain):
+        # Redefine exception here but use the right base exc. This is to improve tolerance against Moto refactors.
+        class InvalidKeyPairNameError(RESTError):
+            code = 400
+            request_id_tag_name = "RequestID"
+            extended_templates = {"custom_response": EC2_ERROR_RESPONSE}
+            env = RESTError.extended_environment(extended_templates)
+
+            def __init__(self, key: Iterable[str]):
+                super().__init__(
+                    "InvalidKeyPair.NotFound",
+                    f"The keypair '{key}' does not exist.",
+                    template="custom_response",
+                )
+
+        # Ensure RESTErrors are translated
+        context = create_aws_request_context(
+            "ec2",
+            "RunInstances",
+            "ec2",
+            {
+                "ImageId": "ami-deadc0de",
+                "InstanceType": "t3.nano",
+                "KeyName": "some-key-pair",
+                "MaxCount": 1,
+                "MinCount": 1,
+            },
+        )
+        moto_exception = InvalidKeyPairNameError({"some-key-pair"})
+
+        ServiceExceptionSerializer().create_exception_response(moto_exception, context)
+
+        assert (
+            "The keypair '{'some-key-pair'}' does not exist." in context.service_exception.message
+        )
+        assert context.service_exception.code == "InvalidKeyPair.NotFound"
+        assert not context.service_exception.sender_fault
+        assert context.service_exception.status_code == 400