17752573338

Committed 15 Sep 2025 10:56PM UTC coverage: 86.879% (+0.03%) from 86.851%

Build # 17752573338

Build Type

push

github

Committed by

web-flow

Commit Message

CFn: validate during get template (#13139)

Coverage Stats

6 of 6 new or added lines in 1 file covered. (100.0%)

138 existing lines in 10 files now uncovered.

67201 of 77350 relevant lines covered (86.88%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.73

/localstack-core/localstack/services/moto.py

"""
This module provides tools to call Moto service implementations.
"""

import copy
import sys
from collections.abc import Callable
from contextlib import AbstractContextManager
from functools import lru_cache

import moto.backends as moto_backends
from moto.core.base_backend import BackendDict
from moto.core.exceptions import RESTError, ServiceException
from rolo.router import RegexConverter
from werkzeug.exceptions import NotFound
from werkzeug.routing import Map, Rule

from localstack import constants
from localstack.aws.api import (
    CommonServiceException,
    RequestContext,
    ServiceRequest,
    ServiceResponse,
)
from localstack.aws.forwarder import (
    ForwardingFallbackDispatcher,
    create_aws_request_context,
    dispatch_to_backend,
)
from localstack.aws.skeleton import DispatchTable
from localstack.constants import DEFAULT_AWS_ACCOUNT_ID
from localstack.constants import VERSION as LOCALSTACK_VERSION
from localstack.http import Response
from localstack.http.request import Request, get_full_raw_path, get_raw_current_url

MotoDispatcher = Callable[[Request, str, dict], Response]

user_agent = f"Localstack/{LOCALSTACK_VERSION} Python/{sys.version.split(' ')[0]}"


def call_moto(context: RequestContext, include_response_metadata=False) -> ServiceResponse:
    """
    Call moto with the given request context and receive a parsed ServiceResponse.

    :param context: the request context
    :param include_response_metadata: whether to include botocore's "ResponseMetadata" attribute
    :return: a serialized AWS ServiceResponse (same as boto3 would return)
    """
    return dispatch_to_backend(context, dispatch_to_moto, include_response_metadata)


def call_moto_with_request(
    context: RequestContext, service_request: ServiceRequest
) -> ServiceResponse:
    """
    Like `call_moto`, but you can pass a modified version of the service request before calling moto. The caveat is
    that a new HTTP request has to be created. The service_request is serialized into a new RequestContext object,
    and headers from the old request are merged into the new one.

    :param context: the original request context
    :param service_request: the dictionary containing the service request parameters
    :return: an ASF ServiceResponse (same as a service provider would return)
    """
    local_context = create_aws_request_context(
        service_name=context.service.service_name,
        action=context.operation.name,
        parameters=service_request,
        region=context.region,
    )
    # we keep the headers from the original request, but override them with the ones created from the `service_request`
    headers = copy.deepcopy(context.request.headers)
    headers.update(local_context.request.headers)
    local_context.request.headers = headers

    return call_moto(local_context)


def _proxy_moto(
    context: RequestContext, request: ServiceRequest
) -> ServiceResponse | Response | None:
    """
    Wraps `call_moto` such that the interface is compliant with a ServiceRequestHandler.

    :param context: the request context
    :param service_request: currently not being used, added to satisfy ServiceRequestHandler contract
    :return: the Response from moto
    """
    return call_moto(context)


def MotoFallbackDispatcher(provider: object) -> DispatchTable:
    """
    Wraps a provider with a moto fallthrough mechanism. It does by creating a new DispatchTable from the original
    provider, and wrapping each method with a fallthrough method that calls ``request`` if the original provider
    raises a ``NotImplementedError``.

    :param provider: the ASF provider
    :return: a modified DispatchTable
    """
    return ForwardingFallbackDispatcher(provider, _proxy_moto)


def dispatch_to_moto(context: RequestContext) -> Response:
    """
    Internal method to dispatch the request to moto without changing moto's dispatcher output.
    :param context: the request context
    :return: the response from moto
    """
    service = context.service
    request = context.request

    # Werkzeug might have an issue (to be determined where the responsibility lies) with proxied requests where the
    # HTTP location is a full URI and not only a path.
    # We need to use the full_raw_url as moto does some path decoding (in S3 for example)
    full_raw_path = get_full_raw_path(request)
    # remove the query string from the full path to do the matching of the request
    raw_path_only = full_raw_path.split("?")[0]
    # this is where we skip the HTTP roundtrip between the moto server and the boto client
    dispatch = get_dispatcher(service.service_name, raw_path_only)
    try:
        raw_url = get_raw_current_url(
            request.scheme, request.host, request.root_path, full_raw_path
        )
        response = dispatch(request, raw_url, request.headers)
        if not response:
            # some operations are only partially implemented by moto
            # e.g. the request will be resolved, but then the request method is not handled
            # it will return None in that case, e.g. for: apigateway TestInvokeAuthorizer + UpdateGatewayResponse
            raise NotImplementedError
        status, headers, content = response
        if isinstance(content, str) and len(content) == 0:
            # moto often returns an empty string to indicate an empty body.
            # use None instead to ensure that body-related headers aren't overwritten when creating the response object.
            content = None
        return Response(content, status, headers)
    except RESTError as e:
        raise CommonServiceException(e.error_type, e.message, status_code=e.code) from e


def get_dispatcher(service: str, path: str) -> MotoDispatcher:
    url_map = get_moto_routing_table(service)

    if len(url_map._rules) == 1:
        # in most cases, there will only be one dispatch method in the list of urls, so no need to do matching
        rule = next(url_map.iter_rules())
        return rule.endpoint

    matcher = url_map.bind(constants.LOCALHOST)
    try:
        endpoint, _ = matcher.match(path_info=path)
    except NotFound as e:
        raise NotImplementedError(
            f"No moto route for service {service} on path {path} found."
        ) from e
    return endpoint


@lru_cache
def get_moto_routing_table(service: str) -> Map:
    """Cached version of load_moto_routing_table."""
    return load_moto_routing_table(service)


def load_moto_routing_table(service: str) -> Map:
    """
    Creates from moto service url_paths a werkzeug URL rule map that can be used to locate moto methods to dispatch
    requests to.

    :param service: the service to get the map for.
    :return: a new Map object
    """
    # code from moto.moto_server.werkzeug_app.create_backend_app
    backend_dict = moto_backends.get_backend(service)
    # Get an instance of this backend.
    # We'll only use this backend to resolve the URL's, so the exact region/account_id is irrelevant
    if isinstance(backend_dict, BackendDict):
        if "us-east-1" in backend_dict[DEFAULT_AWS_ACCOUNT_ID]:
            backend = backend_dict[DEFAULT_AWS_ACCOUNT_ID]["us-east-1"]
        else:
            backend = backend_dict[DEFAULT_AWS_ACCOUNT_ID]["global"]
    else:
        backend = backend_dict["global"]

    url_map = Map()
    url_map.converters["regex"] = _PartIsolatingRegexConverter

    for url_path, handler in backend.flask_paths.items():
        # Some URL patterns in moto have optional trailing slashes, for example the route53 pattern:
        # r"{0}/(?P<api_version>[\d_-]+)/hostedzone/(?P<zone_id>[^/]+)/rrset/?$".
        # However, they don't actually seem to work. Routing only works because moto disables strict_slashes check
        # for the URL Map. So we also disable it here explicitly.
        strict_slashes = False

        # Rule endpoints are annotated as string types in werkzeug, but they don't have to be.
        endpoint = handler

        url_map.add(Rule(url_path, endpoint=endpoint, strict_slashes=strict_slashes))

    return url_map


class _PartIsolatingRegexConverter(RegexConverter):
    """
    Werkzeug converter with disabled path isolation.
    This converter is equivalent to moto.moto_server.utilities.RegexConverter.
    It is necessary to be duplicated here to avoid a transitive import of flask.
    """

    part_isolating = False

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)


class ServiceExceptionTranslator(AbstractContextManager):
    """
    This reentrant context manager translates Moto exceptions into ASF service exceptions. This allows ASF to properly
    serialise and generate the correct error response.

    This is useful when invoking Moto operations directly by importing the backend. For example:

        from moto.ses import ses_backends

        backend = ses_backend['000000000000']['us-east-1']

        with ServiceExceptionTranslator():
            message = backend.send_raw_email(...)

    If `send_raw_email(...)` raises any `moto.core.exceptions.ServiceException`, this context manager will transparently
    generate and raise a `localstack.aws.api.core.CommonServiceException`, maintaining the error code and message.

    This only works for Moto services that are integrated with its new core AWS response serialiser.
    """

    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None and issubclass(exc_type, ServiceException):
            raise CommonServiceException(
                code=exc_val.code,
                message=exc_val.message,
            )
        return False


translate_service_exception = ServiceExceptionTranslator()

1	"""
2	This module provides tools to call Moto service implementations.
3	"""
4
5	import copy	1✔
6	import sys	1✔
7	from collections.abc import Callable	1✔
8	from contextlib import AbstractContextManager	1✔
9	from functools import lru_cache	1✔
10
11	import moto.backends as moto_backends	1✔
12	from moto.core.base_backend import BackendDict	1✔
13	from moto.core.exceptions import RESTError, ServiceException	1✔
14	from rolo.router import RegexConverter	1✔
15	from werkzeug.exceptions import NotFound	1✔
16	from werkzeug.routing import Map, Rule	1✔
17
18	from localstack import constants	1✔
19	from localstack.aws.api import (	1✔
20	CommonServiceException,
21	RequestContext,
22	ServiceRequest,
23	ServiceResponse,
24	)
25	from localstack.aws.forwarder import (	1✔
26	ForwardingFallbackDispatcher,
27	create_aws_request_context,
28	dispatch_to_backend,
29	)
30	from localstack.aws.skeleton import DispatchTable	1✔
31	from localstack.constants import DEFAULT_AWS_ACCOUNT_ID	1✔
32	from localstack.constants import VERSION as LOCALSTACK_VERSION	1✔
33	from localstack.http import Response	1✔
34	from localstack.http.request import Request, get_full_raw_path, get_raw_current_url	1✔
35
36	MotoDispatcher = Callable[[Request, str, dict], Response]	1✔
37
38	user_agent = f"Localstack/{LOCALSTACK_VERSION} Python/{sys.version.split(' ')[0]}"	1✔
39
40
41	def call_moto(context: RequestContext, include_response_metadata=False) -> ServiceResponse:	1✔
42	"""
43	Call moto with the given request context and receive a parsed ServiceResponse.
44
45	:param context: the request context
46	:param include_response_metadata: whether to include botocore's "ResponseMetadata" attribute
47	:return: a serialized AWS ServiceResponse (same as boto3 would return)
48	"""
49	return dispatch_to_backend(context, dispatch_to_moto, include_response_metadata)	1✔
50
51
52	def call_moto_with_request(	1✔
53	context: RequestContext, service_request: ServiceRequest
54	) -> ServiceResponse:
55	"""
56	Like `call_moto`, but you can pass a modified version of the service request before calling moto. The caveat is
57	that a new HTTP request has to be created. The service_request is serialized into a new RequestContext object,
58	and headers from the old request are merged into the new one.
59
60	:param context: the original request context
61	:param service_request: the dictionary containing the service request parameters
62	:return: an ASF ServiceResponse (same as a service provider would return)
63	"""
64	local_context = create_aws_request_context(	1✔
65	service_name=context.service.service_name,
66	action=context.operation.name,
67	parameters=service_request,
68	region=context.region,
69	)
70	# we keep the headers from the original request, but override them with the ones created from the `service_request`
71	headers = copy.deepcopy(context.request.headers)	1✔
72	headers.update(local_context.request.headers)	1✔
73	local_context.request.headers = headers	1✔
74
75	return call_moto(local_context)	1✔
76
77
78	def _proxy_moto(	1✔
79	context: RequestContext, request: ServiceRequest
80	) -> ServiceResponse \| Response \| None:
81	"""
82	Wraps `call_moto` such that the interface is compliant with a ServiceRequestHandler.
83
84	:param context: the request context
85	:param service_request: currently not being used, added to satisfy ServiceRequestHandler contract
86	:return: the Response from moto
87	"""
88	return call_moto(context)	1✔
89
90
91	def MotoFallbackDispatcher(provider: object) -> DispatchTable:	1✔
92	"""
93	Wraps a provider with a moto fallthrough mechanism. It does by creating a new DispatchTable from the original
94	provider, and wrapping each method with a fallthrough method that calls ``request`` if the original provider
95	raises a ``NotImplementedError``.
96
97	:param provider: the ASF provider
98	:return: a modified DispatchTable
99	"""
100	return ForwardingFallbackDispatcher(provider, _proxy_moto)	1✔
101
102
103	def dispatch_to_moto(context: RequestContext) -> Response:	1✔
104	"""
105	Internal method to dispatch the request to moto without changing moto's dispatcher output.
106	:param context: the request context
107	:return: the response from moto
108	"""
109	service = context.service	1✔
110	request = context.request	1✔
111
112	# Werkzeug might have an issue (to be determined where the responsibility lies) with proxied requests where the
113	# HTTP location is a full URI and not only a path.
114	# We need to use the full_raw_url as moto does some path decoding (in S3 for example)
115	full_raw_path = get_full_raw_path(request)	1✔
116	# remove the query string from the full path to do the matching of the request
117	raw_path_only = full_raw_path.split("?")[0]	1✔
118	# this is where we skip the HTTP roundtrip between the moto server and the boto client
119	dispatch = get_dispatcher(service.service_name, raw_path_only)	1✔
120	try:	1✔
121	raw_url = get_raw_current_url(	1✔
122	request.scheme, request.host, request.root_path, full_raw_path
123	)
124	response = dispatch(request, raw_url, request.headers)	1✔
125	if not response:	1✔
126	# some operations are only partially implemented by moto
127	# e.g. the request will be resolved, but then the request method is not handled
128	# it will return None in that case, e.g. for: apigateway TestInvokeAuthorizer + UpdateGatewayResponse
129	raise NotImplementedError
130	status, headers, content = response	1✔
131	if isinstance(content, str) and len(content) == 0:	1✔
132	# moto often returns an empty string to indicate an empty body.
133	# use None instead to ensure that body-related headers aren't overwritten when creating the response object.
134	content = None	1✔
135	return Response(content, status, headers)	1✔
136	except RESTError as e:	1✔
UNCOV 137	raise CommonServiceException(e.error_type, e.message, status_code=e.code) from e	×
138
139
140	def get_dispatcher(service: str, path: str) -> MotoDispatcher:	1✔
141	url_map = get_moto_routing_table(service)	1✔
142
143	if len(url_map._rules) == 1:	1✔
144	# in most cases, there will only be one dispatch method in the list of urls, so no need to do matching
145	rule = next(url_map.iter_rules())	1✔
146	return rule.endpoint	1✔
147
148	matcher = url_map.bind(constants.LOCALHOST)	1✔
149	try:	1✔
150	endpoint, _ = matcher.match(path_info=path)	1✔
151	except NotFound as e:	1✔
152	raise NotImplementedError(
153	f"No moto route for service {service} on path {path} found."
154	) from e
155	return endpoint	1✔
156
157
158	@lru_cache	1✔
159	def get_moto_routing_table(service: str) -> Map:	1✔
160	"""Cached version of load_moto_routing_table."""
161	return load_moto_routing_table(service)	1✔
162
163
164	def load_moto_routing_table(service: str) -> Map:	1✔
165	"""
166	Creates from moto service url_paths a werkzeug URL rule map that can be used to locate moto methods to dispatch
167	requests to.
168
169	:param service: the service to get the map for.
170	:return: a new Map object
171	"""
172	# code from moto.moto_server.werkzeug_app.create_backend_app
173	backend_dict = moto_backends.get_backend(service)	1✔
174	# Get an instance of this backend.
175	# We'll only use this backend to resolve the URL's, so the exact region/account_id is irrelevant
176	if isinstance(backend_dict, BackendDict):	1✔
177	if "us-east-1" in backend_dict[DEFAULT_AWS_ACCOUNT_ID]:	1✔
178	backend = backend_dict[DEFAULT_AWS_ACCOUNT_ID]["us-east-1"]	1✔
179	else:
180	backend = backend_dict[DEFAULT_AWS_ACCOUNT_ID]["global"]	1✔
181	else:
UNCOV 182	backend = backend_dict["global"]	×
183
184	url_map = Map()	1✔
185	url_map.converters["regex"] = _PartIsolatingRegexConverter	1✔
186
187	for url_path, handler in backend.flask_paths.items():	1✔
188	# Some URL patterns in moto have optional trailing slashes, for example the route53 pattern:
189	# r"{0}/(?P<api_version>[\d_-]+)/hostedzone/(?P<zone_id>[^/]+)/rrset/?$".
190	# However, they don't actually seem to work. Routing only works because moto disables strict_slashes check
191	# for the URL Map. So we also disable it here explicitly.
192	strict_slashes = False	1✔
193
194	# Rule endpoints are annotated as string types in werkzeug, but they don't have to be.
195	endpoint = handler	1✔
196
197	url_map.add(Rule(url_path, endpoint=endpoint, strict_slashes=strict_slashes))	1✔
198
199	return url_map	1✔
200
201
202	class _PartIsolatingRegexConverter(RegexConverter):	1✔
203	"""
204	Werkzeug converter with disabled path isolation.
205	This converter is equivalent to moto.moto_server.utilities.RegexConverter.
206	It is necessary to be duplicated here to avoid a transitive import of flask.
207	"""
208
209	part_isolating = False	1✔
210
211	def __init__(self, args, *kwargs) -> None:	1✔
212	super().__init__(args, *kwargs)	1✔
213
214
215	class ServiceExceptionTranslator(AbstractContextManager):	1✔
216	"""
217	This reentrant context manager translates Moto exceptions into ASF service exceptions. This allows ASF to properly
218	serialise and generate the correct error response.
219
220	This is useful when invoking Moto operations directly by importing the backend. For example:
221
222	from moto.ses import ses_backends
223
224	backend = ses_backend['000000000000']['us-east-1']
225
226	with ServiceExceptionTranslator():
227	message = backend.send_raw_email(...)
228
229	If `send_raw_email(...)` raises any `moto.core.exceptions.ServiceException`, this context manager will transparently
230	generate and raise a `localstack.aws.api.core.CommonServiceException`, maintaining the error code and message.
231
232	This only works for Moto services that are integrated with its new core AWS response serialiser.
233	"""
234
235	def __enter__(self):	1✔
236	pass	1✔
237
238	def __exit__(self, exc_type, exc_val, exc_tb):	1✔
239	if exc_type is not None and issubclass(exc_type, ServiceException):	1✔
240	raise CommonServiceException(	1✔
241	code=exc_val.code,
242	message=exc_val.message,
243	)
244	return False	1✔
245
246
247	translate_service_exception = ServiceExceptionTranslator()	1✔

localstack / localstack / 17752573338

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous