16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

98.89

/localstack-core/localstack/aws/forwarder.py

"""
This module contains utilities to call a backend (e.g., an external service process like
DynamoDBLocal) from a service provider.
"""

from collections.abc import Callable, Mapping
from typing import Any

from botocore.awsrequest import AWSPreparedRequest, prepare_request_dict
from botocore.config import Config as BotoConfig
from werkzeug.datastructures import Headers

from localstack.aws.api.core import (
    RequestContext,
    ServiceRequest,
    ServiceRequestHandler,
    ServiceResponse,
)
from localstack.aws.client import create_http_request, parse_response, raise_service_exception
from localstack.aws.connect import connect_to
from localstack.aws.skeleton import DispatchTable, create_dispatch_table
from localstack.aws.spec import load_service
from localstack.constants import AWS_REGION_US_EAST_1
from localstack.http import Response
from localstack.http.proxy import Proxy


class AwsRequestProxy:
    """
    Implements the ``ServiceRequestHandler`` protocol to forward AWS requests to a backend. It is stateful and uses a
    ``Proxy`` instance for re-using client connections to the backend.
    """

    def __init__(
        self,
        endpoint_url: str,
        parse_response: bool = True,
        include_response_metadata: bool = False,
    ):
        """
        Create a new AwsRequestProxy. ``parse_response`` control the return behavior of ``forward``. If
        ``parse_response`` is set, then ``forward`` parses the HTTP response from the backend and returns a
        ``ServiceResponse``, otherwise it returns the raw HTTP ``Response`` object.

        :param endpoint_url: the backend to proxy the requests to, used as ``forward_base_url`` for the ``Proxy``.
        :param parse_response: whether to parse the response before returning it
        :param include_response_metadata: include AWS response metadata, only used with ``parse_response=True``
        """
        self.endpoint_url = endpoint_url
        self.parse_response = parse_response
        self.include_response_metadata = include_response_metadata
        self.proxy = Proxy(forward_base_url=endpoint_url)

    def __call__(
        self,
        context: RequestContext,
        service_request: ServiceRequest = None,
    ) -> ServiceResponse | Response | None:
        """Method to satisfy the ``ServiceRequestHandler`` protocol."""
        return self.forward(context, service_request)

    def forward(
        self,
        context: RequestContext,
        service_request: ServiceRequest = None,
    ) -> ServiceResponse | Response | None:
        """
        Forwards the given request to the backend configured by ``endpoint_url``.

        :param context: the original request context of the incoming request
        :param service_request: optionally a new service
        :return:
        """
        if service_request is not None:
            # if a service request is passed then we need to create a new request context
            context = self.new_request_context(context, service_request)

        http_response = self.proxy.forward(context.request, forward_path=context.request.path)
        if not self.parse_response:
            return http_response
        parsed_response = parse_response(
            context.operation, http_response, self.include_response_metadata
        )
        raise_service_exception(http_response, parsed_response)
        return parsed_response

    def new_request_context(self, original: RequestContext, service_request: ServiceRequest):
        context = create_aws_request_context(
            service_name=original.service.service_name,
            action=original.operation.name,
            parameters=service_request,
            region=original.region,
        )
        # update the newly created context with non-payload specific request headers (the payload can differ from
        # the original request, f.e. it could be JSON encoded now while the initial request was CBOR encoded)
        headers = Headers(original.request.headers)
        headers.pop("Content-Type", None)
        headers.pop("Content-Length", None)
        context.request.headers.update(headers)
        return context


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

    :param provider: the ASF provider
    :param request_forwarder: callable that forwards the request (e.g., to a backend server)
    :return: a modified DispatchTable
    """
    table = create_dispatch_table(provider)

    for op, fn in table.items():
        table[op] = _wrap_with_fallthrough(fn, request_forwarder)

    return table


class NotImplementedAvoidFallbackError(NotImplementedError):
    pass


def _wrap_with_fallthrough(
    handler: ServiceRequestHandler, fallthrough_handler: ServiceRequestHandler
) -> ServiceRequestHandler:
    def _call(context, req) -> ServiceResponse:
        try:
            # handler will typically be an ASF provider method, and in case it hasn't been
            # implemented, we try to fall back to forwarding the request to the backend
            return handler(context, req)
        except NotImplementedAvoidFallbackError as e:
            # if the fallback has been explicitly disabled, don't pass on to the fallback
            raise e
        except NotImplementedError:
            pass

        return fallthrough_handler(context, req)

    return _call


def HttpFallbackDispatcher(provider: object, forward_url_getter: Callable[[str, str], str]):
    return ForwardingFallbackDispatcher(provider, get_request_forwarder_http(forward_url_getter))


def get_request_forwarder_http(
    forward_url_getter: Callable[[str, str], str],
) -> ServiceRequestHandler:
    """
    Returns a ServiceRequestHandler that creates for each invocation a new AwsRequestProxy with the result of
    forward_url_getter. Note that this is an inefficient method of proxying, since for every call a new client
    connection has to be established. Try to instead use static forward URL values and use ``AwsRequestProxy`` directly.

    :param forward_url_getter: a factory method for returning forward base urls for the proxy
    :return: a ServiceRequestHandler acting as a proxy
    """

    def _forward_request(
        context: RequestContext, service_request: ServiceRequest = None
    ) -> ServiceResponse:
        return AwsRequestProxy(forward_url_getter(context.account_id, context.region)).forward(
            context, service_request
        )

    return _forward_request


def dispatch_to_backend(
    context: RequestContext,
    http_request_dispatcher: Callable[[RequestContext], Response],
    include_response_metadata=False,
) -> ServiceResponse:
    """
    Dispatch the given request to a backend by using the `request_forwarder` function to
    fetch an HTTP response, converting it to a ServiceResponse.
    :param context: the request context
    :param http_request_dispatcher: dispatcher that performs the request and returns an HTTP response
    :param include_response_metadata: whether to include boto3 response metadata in the response
    :return: parsed service response
    :raises ServiceException: if the dispatcher returned an error response
    """
    http_response = http_request_dispatcher(context)
    parsed_response = parse_response(context.operation, http_response, include_response_metadata)
    raise_service_exception(http_response, parsed_response)
    return parsed_response


# boto config deactivating param validation to forward to backends (backends are responsible for validating params)
_non_validating_boto_config = BotoConfig(parameter_validation=False)


def create_aws_request_context(
    service_name: str,
    action: str,
    parameters: Mapping[str, Any] = None,
    region: str = None,
    endpoint_url: str | None = None,
) -> RequestContext:
    """
    This is a stripped-down version of what the botocore client does to perform an HTTP request from a client call. A
    client call looks something like this: boto3.client("sqs").create_queue(QueueName="myqueue"), which will be
    serialized into an HTTP request. This method does the same, without performing the actual request, and with a
    more low-level interface. An equivalent call would be

         create_aws_request_context("sqs", "CreateQueue", {"QueueName": "myqueue"})

    :param service_name: the AWS service
    :param action: the action to invoke
    :param parameters: the invocation parameters
    :param region: the region name (default is us-east-1)
    :param endpoint_url: the endpoint to call (defaults to localstack)
    :return: a RequestContext object that describes this request
    """
    if parameters is None:
        parameters = {}
    if region is None:
        region = AWS_REGION_US_EAST_1

    service = load_service(service_name)
    operation = service.operation_model(action)

    # we re-use botocore internals here to serialize the HTTP request,
    # but deactivate validation (validation errors should be handled by the backend)
    # and don't send it yet
    client = connect_to.get_client(
        service_name,
        endpoint_url=endpoint_url,
        region_name=region,
        config=_non_validating_boto_config,
    )
    request_context = {
        "client_region": region,
        "has_streaming_input": operation.has_streaming_input,
        "auth_type": operation.auth_type,
    }

    # The endpoint URL is mandatory here, set a dummy if not given (doesn't _need_ to be localstack specific)
    if not endpoint_url:
        endpoint_url = "http://localhost.localstack.cloud"
    # pre-process the request args (some params are modified using botocore event handlers)
    parameters = client._emit_api_params(parameters, operation, request_context)
    request_dict = client._convert_to_request_dict(
        parameters, operation, endpoint_url, context=request_context
    )

    if auth_path := request_dict.get("auth_path"):
        # botocore >= 1.28 might modify the url path of the request dict (specifically for S3).
        # It will then set the original url path as "auth_path". If the auth_path is set, we reset the url_path.
        # Since botocore 1.31.2, botocore will strip the query from the `authPart`
        # We need to add it back from `requestUri` field
        # Afterwards the request needs to be prepared again.
        path, sep, query = request_dict["url_path"].partition("?")
        request_dict["url_path"] = f"{auth_path}{sep}{query}"
        prepare_request_dict(
            request_dict,
            endpoint_url=endpoint_url,
            user_agent=client._client_config.user_agent,
            context=request_context,
        )

    aws_request: AWSPreparedRequest = client._endpoint.create_request(request_dict, operation)
    context = RequestContext(request=create_http_request(aws_request))
    context.service = service
    context.operation = operation
    context.region = region
    context.service_request = parameters

    return context

1	"""
2	This module contains utilities to call a backend (e.g., an external service process like
3	DynamoDBLocal) from a service provider.
4	"""
5
6	from collections.abc import Callable, Mapping	1✔
7	from typing import Any	1✔
8
9	from botocore.awsrequest import AWSPreparedRequest, prepare_request_dict	1✔
10	from botocore.config import Config as BotoConfig	1✔
11	from werkzeug.datastructures import Headers	1✔
12
13	from localstack.aws.api.core import (	1✔
14	RequestContext,
15	ServiceRequest,
16	ServiceRequestHandler,
17	ServiceResponse,
18	)
19	from localstack.aws.client import create_http_request, parse_response, raise_service_exception	1✔
20	from localstack.aws.connect import connect_to	1✔
21	from localstack.aws.skeleton import DispatchTable, create_dispatch_table	1✔
22	from localstack.aws.spec import load_service	1✔
23	from localstack.constants import AWS_REGION_US_EAST_1	1✔
24	from localstack.http import Response	1✔
25	from localstack.http.proxy import Proxy	1✔
26
27
28	class AwsRequestProxy:	1✔
29	"""
30	Implements the ``ServiceRequestHandler`` protocol to forward AWS requests to a backend. It is stateful and uses a
31	``Proxy`` instance for re-using client connections to the backend.
32	"""
33
34	def __init__(	1✔
35	self,
36	endpoint_url: str,
37	parse_response: bool = True,
38	include_response_metadata: bool = False,
39	):
40	"""
41	Create a new AwsRequestProxy. ``parse_response`` control the return behavior of ``forward``. If
42	``parse_response`` is set, then ``forward`` parses the HTTP response from the backend and returns a
43	``ServiceResponse``, otherwise it returns the raw HTTP ``Response`` object.
44
45	:param endpoint_url: the backend to proxy the requests to, used as ``forward_base_url`` for the ``Proxy``.
46	:param parse_response: whether to parse the response before returning it
47	:param include_response_metadata: include AWS response metadata, only used with ``parse_response=True``
48	"""
49	self.endpoint_url = endpoint_url	1✔
50	self.parse_response = parse_response	1✔
51	self.include_response_metadata = include_response_metadata	1✔
52	self.proxy = Proxy(forward_base_url=endpoint_url)	1✔
53
54	def __call__(	1✔
55	self,
56	context: RequestContext,
57	service_request: ServiceRequest = None,
58	) -> ServiceResponse \| Response \| None:
59	"""Method to satisfy the ``ServiceRequestHandler`` protocol."""
60	return self.forward(context, service_request)	1✔
61
62	def forward(	1✔
63	self,
64	context: RequestContext,
65	service_request: ServiceRequest = None,
66	) -> ServiceResponse \| Response \| None:
67	"""
68	Forwards the given request to the backend configured by ``endpoint_url``.
69
70	:param context: the original request context of the incoming request
71	:param service_request: optionally a new service
72	:return:
73	"""
74	if service_request is not None:	1✔
75	# if a service request is passed then we need to create a new request context
76	context = self.new_request_context(context, service_request)	1✔
77
78	http_response = self.proxy.forward(context.request, forward_path=context.request.path)	1✔
79	if not self.parse_response:	1✔
UNCOV 80	return http_response	×
81	parsed_response = parse_response(	1✔
82	context.operation, http_response, self.include_response_metadata
83	)
84	raise_service_exception(http_response, parsed_response)	1✔
85	return parsed_response	1✔
86
87	def new_request_context(self, original: RequestContext, service_request: ServiceRequest):	1✔
88	context = create_aws_request_context(	1✔
89	service_name=original.service.service_name,
90	action=original.operation.name,
91	parameters=service_request,
92	region=original.region,
93	)
94	# update the newly created context with non-payload specific request headers (the payload can differ from
95	# the original request, f.e. it could be JSON encoded now while the initial request was CBOR encoded)
96	headers = Headers(original.request.headers)	1✔
97	headers.pop("Content-Type", None)	1✔
98	headers.pop("Content-Length", None)	1✔
99	context.request.headers.update(headers)	1✔
100	return context	1✔
101
102
103	def ForwardingFallbackDispatcher(	1✔
104	provider: object, request_forwarder: ServiceRequestHandler
105	) -> DispatchTable:
106	"""
107	Wraps a provider with a request forwarder. It does by creating a new DispatchTable from the original
108	provider, and wrapping each method with a fallthrough method that calls ``request_forwarder`` if the
109	original provider raises a ``NotImplementedError``.
110
111	:param provider: the ASF provider
112	:param request_forwarder: callable that forwards the request (e.g., to a backend server)
113	:return: a modified DispatchTable
114	"""
115	table = create_dispatch_table(provider)	1✔
116
117	for op, fn in table.items():	1✔
118	table[op] = _wrap_with_fallthrough(fn, request_forwarder)	1✔
119
120	return table	1✔
121
122
123	class NotImplementedAvoidFallbackError(NotImplementedError):	1✔
124	pass	1✔
125
126
127	def _wrap_with_fallthrough(	1✔
128	handler: ServiceRequestHandler, fallthrough_handler: ServiceRequestHandler
129	) -> ServiceRequestHandler:
130	def _call(context, req) -> ServiceResponse:	1✔
131	try:	1✔
132	# handler will typically be an ASF provider method, and in case it hasn't been
133	# implemented, we try to fall back to forwarding the request to the backend
134	return handler(context, req)	1✔
135	except NotImplementedAvoidFallbackError as e:	1✔
136	# if the fallback has been explicitly disabled, don't pass on to the fallback
137	raise e	1✔
138	except NotImplementedError:	1✔
139	pass	1✔
140
141	return fallthrough_handler(context, req)	1✔
142
143	return _call	1✔
144
145
146	def HttpFallbackDispatcher(provider: object, forward_url_getter: Callable[[str, str], str]):	1✔
147	return ForwardingFallbackDispatcher(provider, get_request_forwarder_http(forward_url_getter))	1✔
148
149
150	def get_request_forwarder_http(	1✔
151	forward_url_getter: Callable[[str, str], str],
152	) -> ServiceRequestHandler:
153	"""
154	Returns a ServiceRequestHandler that creates for each invocation a new AwsRequestProxy with the result of
155	forward_url_getter. Note that this is an inefficient method of proxying, since for every call a new client
156	connection has to be established. Try to instead use static forward URL values and use ``AwsRequestProxy`` directly.
157
158	:param forward_url_getter: a factory method for returning forward base urls for the proxy
159	:return: a ServiceRequestHandler acting as a proxy
160	"""
161
162	def _forward_request(	1✔
163	context: RequestContext, service_request: ServiceRequest = None
164	) -> ServiceResponse:
165	return AwsRequestProxy(forward_url_getter(context.account_id, context.region)).forward(	1✔
166	context, service_request
167	)
168
169	return _forward_request	1✔
170
171
172	def dispatch_to_backend(	1✔
173	context: RequestContext,
174	http_request_dispatcher: Callable[[RequestContext], Response],
175	include_response_metadata=False,
176	) -> ServiceResponse:
177	"""
178	Dispatch the given request to a backend by using the `request_forwarder` function to
179	fetch an HTTP response, converting it to a ServiceResponse.
180	:param context: the request context
181	:param http_request_dispatcher: dispatcher that performs the request and returns an HTTP response
182	:param include_response_metadata: whether to include boto3 response metadata in the response
183	:return: parsed service response
184	:raises ServiceException: if the dispatcher returned an error response
185	"""
186	http_response = http_request_dispatcher(context)	1✔
187	parsed_response = parse_response(context.operation, http_response, include_response_metadata)	1✔
188	raise_service_exception(http_response, parsed_response)	1✔
189	return parsed_response	1✔
190
191
192	# boto config deactivating param validation to forward to backends (backends are responsible for validating params)
193	_non_validating_boto_config = BotoConfig(parameter_validation=False)	1✔
194
195
196	def create_aws_request_context(	1✔
197	service_name: str,
198	action: str,
199	parameters: Mapping[str, Any] = None,
200	region: str = None,
201	endpoint_url: str \| None = None,
202	) -> RequestContext:
203	"""
204	This is a stripped-down version of what the botocore client does to perform an HTTP request from a client call. A
205	client call looks something like this: boto3.client("sqs").create_queue(QueueName="myqueue"), which will be
206	serialized into an HTTP request. This method does the same, without performing the actual request, and with a
207	more low-level interface. An equivalent call would be
208
209	create_aws_request_context("sqs", "CreateQueue", {"QueueName": "myqueue"})
210
211	:param service_name: the AWS service
212	:param action: the action to invoke
213	:param parameters: the invocation parameters
214	:param region: the region name (default is us-east-1)
215	:param endpoint_url: the endpoint to call (defaults to localstack)
216	:return: a RequestContext object that describes this request
217	"""
218	if parameters is None:	1✔
219	parameters = {}	1✔
220	if region is None:	1✔
221	region = AWS_REGION_US_EAST_1	1✔
222
223	service = load_service(service_name)	1✔
224	operation = service.operation_model(action)	1✔
225
226	# we re-use botocore internals here to serialize the HTTP request,
227	# but deactivate validation (validation errors should be handled by the backend)
228	# and don't send it yet
229	client = connect_to.get_client(	1✔
230	service_name,
231	endpoint_url=endpoint_url,
232	region_name=region,
233	config=_non_validating_boto_config,
234	)
235	request_context = {	1✔
236	"client_region": region,
237	"has_streaming_input": operation.has_streaming_input,
238	"auth_type": operation.auth_type,
239	}
240
241	# The endpoint URL is mandatory here, set a dummy if not given (doesn't _need_ to be localstack specific)
242	if not endpoint_url:	1✔
243	endpoint_url = "http://localhost.localstack.cloud"	1✔
244	# pre-process the request args (some params are modified using botocore event handlers)
245	parameters = client._emit_api_params(parameters, operation, request_context)	1✔
246	request_dict = client._convert_to_request_dict(	1✔
247	parameters, operation, endpoint_url, context=request_context
248	)
249
250	if auth_path := request_dict.get("auth_path"):	1✔
251	# botocore >= 1.28 might modify the url path of the request dict (specifically for S3).
252	# It will then set the original url path as "auth_path". If the auth_path is set, we reset the url_path.
253	# Since botocore 1.31.2, botocore will strip the query from the `authPart`
254	# We need to add it back from `requestUri` field
255	# Afterwards the request needs to be prepared again.
256	path, sep, query = request_dict["url_path"].partition("?")	1✔
257	request_dict["url_path"] = f"{auth_path}{sep}{query}"	1✔
258	prepare_request_dict(	1✔
259	request_dict,
260	endpoint_url=endpoint_url,
261	user_agent=client._client_config.user_agent,
262	context=request_context,
263	)
264
265	aws_request: AWSPreparedRequest = client._endpoint.create_request(request_dict, operation)	1✔
266	context = RequestContext(request=create_http_request(aws_request))	1✔
267	context.service = service	1✔
268	context.operation = operation	1✔
269	context.region = region	1✔
270	context.service_request = parameters	1✔
271
272	return context	1✔

localstack / localstack / 16820655284

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