20357994067

Committed 18 Dec 2025 10:01PM UTC coverage: 86.913% (-0.02%) from 86.929%

Build # 20357994067

Build Type

push

github

Committed by

web-flow

Commit Message

Fix CloudWatch model annotation (#13545)

Run Details

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

1391 existing lines in 33 files now uncovered.

70000 of 80540 relevant lines covered (86.91%)

1.72 hits per line

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

96.88

/localstack-core/localstack/services/apigateway/next_gen/execute_api/handlers/parse.py

import datetime
import logging
import re
from collections import defaultdict
from urllib.parse import urlparse

from rolo.request import restore_payload
from werkzeug.datastructures import Headers, MultiDict

from localstack.http import Response
from localstack.services.apigateway.helpers import REQUEST_TIME_DATE_FORMAT
from localstack.utils.strings import long_uid, short_uid
from localstack.utils.time import timestamp

from ..api import RestApiGatewayHandler, RestApiGatewayHandlerChain
from ..context import InvocationRequest, RestApiInvocationContext
from ..header_utils import should_drop_header_from_invocation
from ..helpers import generate_trace_id, generate_trace_parent, parse_trace_id
from ..variables import (
    ContextVariableOverrides,
    ContextVariables,
    ContextVarsIdentity,
    ContextVarsRequestOverride,
    ContextVarsResponseOverride,
)

LOG = logging.getLogger(__name__)


class InvocationRequestParser(RestApiGatewayHandler):
    def __call__(
        self,
        chain: RestApiGatewayHandlerChain,
        context: RestApiInvocationContext,
        response: Response,
    ):
        context.account_id = context.deployment.account_id
        context.region = context.deployment.region
        self.parse_and_enrich(context)

    def parse_and_enrich(self, context: RestApiInvocationContext):
        # first, create the InvocationRequest with the incoming request
        context.invocation_request = self.create_invocation_request(context)
        # then we can create the ContextVariables, used throughout the invocation as payload and to render authorizer
        # payload, mapping templates and such.
        context.context_variables = self.create_context_variables(context)
        context.context_variable_overrides = ContextVariableOverrides(
            requestOverride=ContextVarsRequestOverride(header={}, querystring={}, path={}),
            responseOverride=ContextVarsResponseOverride(header={}, status=0),
        )
        # TODO: maybe adjust the logging
        LOG.debug("Initializing $context='%s'", context.context_variables)
        # then populate the stage variables
        context.stage_variables = self.get_stage_variables(context)
        LOG.debug("Initializing $stageVariables='%s'", context.stage_variables)

        context.trace_id = self.populate_trace_id(context.request.headers)

    def create_invocation_request(self, context: RestApiInvocationContext) -> InvocationRequest:
        request = context.request
        params, multi_value_params = self._get_single_and_multi_values_from_multidict(request.args)
        headers = self._get_invocation_headers(request.headers)
        invocation_request = InvocationRequest(
            http_method=request.method,
            query_string_parameters=params,
            multi_value_query_string_parameters=multi_value_params,
            headers=headers,
            body=restore_payload(request),
        )
        self._enrich_with_raw_path(context, invocation_request)

        return invocation_request

    @staticmethod
    def _enrich_with_raw_path(
        context: RestApiInvocationContext, invocation_request: InvocationRequest
    ):
        # Base path is not URL-decoded, so we need to get the `RAW_URI` from the request
        request = context.request
        raw_uri = request.environ.get("RAW_URI") or request.path

        # if the request comes from the LocalStack only `_user_request_` route, we need to remove this prefix from the
        # path, in order to properly route the request
        if "_user_request_" in raw_uri:
            # in this format, the stage is before `_user_request_`, so we don't need to remove it
            raw_uri = raw_uri.partition("_user_request_")[2]
        else:
            if raw_uri.startswith("/_aws/execute-api"):
                # the API can be cased in the path, so we need to ignore it to remove it
                raw_uri = re.sub(
                    f"^/_aws/execute-api/{context.api_id}",
                    "",
                    raw_uri,
                    flags=re.IGNORECASE,
                )

            # remove the stage from the path, only replace the first occurrence
            raw_uri = raw_uri.replace(f"/{context.stage}", "", 1)

        if raw_uri.startswith("//"):
            # TODO: AWS validate this assumption
            # if the RAW_URI starts with double slashes, `urlparse` will fail to decode it as path only
            # it also means that we already only have the path, so we just need to remove the query string
            raw_uri = raw_uri.split("?")[0]
            raw_path = "/" + raw_uri.lstrip("/")

        else:
            # we need to make sure we have a path here, sometimes RAW_URI can be a full URI (when proxied)
            raw_path = raw_uri = urlparse(raw_uri).path

        invocation_request["path"] = raw_path
        invocation_request["raw_path"] = raw_uri

    @staticmethod
    def _get_single_and_multi_values_from_multidict(
        multi_dict: MultiDict,
    ) -> tuple[dict[str, str], dict[str, list[str]]]:
        single_values = {}
        multi_values = defaultdict(list)

        for key, value in multi_dict.items(multi=True):
            multi_values[key].append(value)
            # for the single value parameters, AWS only keeps the last value of the list
            single_values[key] = value

        return single_values, dict(multi_values)

    @staticmethod
    def _get_invocation_headers(headers: Headers) -> Headers:
        invocation_headers = Headers()
        for key, value in headers:
            if should_drop_header_from_invocation(key):
                LOG.debug("Dropping header from invocation request: '%s'", key)
                continue
            invocation_headers.add(key, value)
        return invocation_headers

    @staticmethod
    def create_context_variables(context: RestApiInvocationContext) -> ContextVariables:
        invocation_request: InvocationRequest = context.invocation_request
        domain_name = invocation_request["headers"].get("Host", "")
        domain_prefix = domain_name.split(".")[0]
        now = datetime.datetime.now()

        context_variables = ContextVariables(
            accountId=context.account_id,
            apiId=context.api_id,
            deploymentId=context.deployment_id,
            domainName=domain_name,
            domainPrefix=domain_prefix,
            extendedRequestId=short_uid(),  # TODO: use snapshot tests to verify format
            httpMethod=invocation_request["http_method"],
            identity=ContextVarsIdentity(
                accountId=None,
                accessKey=None,
                caller=None,
                cognitoAuthenticationProvider=None,
                cognitoAuthenticationType=None,
                cognitoIdentityId=None,
                cognitoIdentityPoolId=None,
                principalOrgId=None,
                sourceIp="127.0.0.1",  # TODO: get the sourceIp from the Request
                user=None,
                userAgent=invocation_request["headers"].get("User-Agent"),
                userArn=None,
            ),
            path=f"/{context.stage}{invocation_request['raw_path']}",
            protocol="HTTP/1.1",
            requestId=long_uid(),
            requestTime=timestamp(time=now, format=REQUEST_TIME_DATE_FORMAT),
            requestTimeEpoch=int(now.timestamp() * 1000),
            stage=context.stage,
        )
        if context.is_canary is not None:
            context_variables["isCanaryRequest"] = context.is_canary

        return context_variables

    @staticmethod
    def get_stage_variables(context: RestApiInvocationContext) -> dict[str, str] | None:
        stage_variables = context.stage_configuration.get("variables")
        if context.is_canary:
            overrides = (
                context.stage_configuration["canarySettings"].get("stageVariableOverrides") or {}
            )
            stage_variables = (stage_variables or {}) | overrides

        if not stage_variables:
            return None

        return stage_variables

    @staticmethod
    def populate_trace_id(headers: Headers) -> str:
        incoming_trace = parse_trace_id(headers.get("x-amzn-trace-id", ""))
        # parse_trace_id always return capitalized keys

        trace = incoming_trace.get("Root", generate_trace_id())
        incoming_parent = incoming_trace.get("Parent")
        parent = incoming_parent or generate_trace_parent()
        sampled = incoming_trace.get("Sampled", "1" if incoming_parent else "0")
        # TODO: lineage? not sure what it related to
        return f"Root={trace};Parent={parent};Sampled={sampled}"

1	import datetime	2✔
2	import logging	2✔
3	import re	2✔
4	from collections import defaultdict	2✔
5	from urllib.parse import urlparse	2✔
6
7	from rolo.request import restore_payload	2✔
8	from werkzeug.datastructures import Headers, MultiDict	2✔
9
10	from localstack.http import Response	2✔
11	from localstack.services.apigateway.helpers import REQUEST_TIME_DATE_FORMAT	2✔
12	from localstack.utils.strings import long_uid, short_uid	2✔
13	from localstack.utils.time import timestamp	2✔
14
15	from ..api import RestApiGatewayHandler, RestApiGatewayHandlerChain	2✔
16	from ..context import InvocationRequest, RestApiInvocationContext	2✔
17	from ..header_utils import should_drop_header_from_invocation	2✔
18	from ..helpers import generate_trace_id, generate_trace_parent, parse_trace_id	2✔
19	from ..variables import (	2✔
20	ContextVariableOverrides,
21	ContextVariables,
22	ContextVarsIdentity,
23	ContextVarsRequestOverride,
24	ContextVarsResponseOverride,
25	)
26
27	LOG = logging.getLogger(__name__)	2✔
28
29
30	class InvocationRequestParser(RestApiGatewayHandler):	2✔
31	def __call__(	2✔
32	self,
33	chain: RestApiGatewayHandlerChain,
34	context: RestApiInvocationContext,
35	response: Response,
36	):
37	context.account_id = context.deployment.account_id	2✔
38	context.region = context.deployment.region	2✔
39	self.parse_and_enrich(context)	2✔
40
41	def parse_and_enrich(self, context: RestApiInvocationContext):	2✔
42	# first, create the InvocationRequest with the incoming request
43	context.invocation_request = self.create_invocation_request(context)	2✔
44	# then we can create the ContextVariables, used throughout the invocation as payload and to render authorizer
45	# payload, mapping templates and such.
46	context.context_variables = self.create_context_variables(context)	2✔
47	context.context_variable_overrides = ContextVariableOverrides(	2✔
48	requestOverride=ContextVarsRequestOverride(header={}, querystring={}, path={}),
49	responseOverride=ContextVarsResponseOverride(header={}, status=0),
50	)
51	# TODO: maybe adjust the logging
52	LOG.debug("Initializing $context='%s'", context.context_variables)	2✔
53	# then populate the stage variables
54	context.stage_variables = self.get_stage_variables(context)	2✔
55	LOG.debug("Initializing $stageVariables='%s'", context.stage_variables)	2✔
56
57	context.trace_id = self.populate_trace_id(context.request.headers)	2✔
58
59	def create_invocation_request(self, context: RestApiInvocationContext) -> InvocationRequest:	2✔
60	request = context.request	2✔
61	params, multi_value_params = self._get_single_and_multi_values_from_multidict(request.args)	2✔
62	headers = self._get_invocation_headers(request.headers)	2✔
63	invocation_request = InvocationRequest(	2✔
64	http_method=request.method,
65	query_string_parameters=params,
66	multi_value_query_string_parameters=multi_value_params,
67	headers=headers,
68	body=restore_payload(request),
69	)
70	self._enrich_with_raw_path(context, invocation_request)	2✔
71
72	return invocation_request	2✔
73
74	@staticmethod	2✔
75	def _enrich_with_raw_path(	2✔
76	context: RestApiInvocationContext, invocation_request: InvocationRequest
77	):
78	# Base path is not URL-decoded, so we need to get the `RAW_URI` from the request
79	request = context.request	2✔
80	raw_uri = request.environ.get("RAW_URI") or request.path	2✔
81
82	# if the request comes from the LocalStack only `_user_request_` route, we need to remove this prefix from the
83	# path, in order to properly route the request
84	if "_user_request_" in raw_uri:	2✔
85	# in this format, the stage is before `_user_request_`, so we don't need to remove it
86	raw_uri = raw_uri.partition("_user_request_")[2]	2✔
87	else:
88	if raw_uri.startswith("/_aws/execute-api"):	2✔
89	# the API can be cased in the path, so we need to ignore it to remove it
90	raw_uri = re.sub(	2✔
91	f"^/_aws/execute-api/{context.api_id}",
92	"",
93	raw_uri,
94	flags=re.IGNORECASE,
95	)
96
97	# remove the stage from the path, only replace the first occurrence
98	raw_uri = raw_uri.replace(f"/{context.stage}", "", 1)	2✔
99
100	if raw_uri.startswith("//"):	2✔
101	# TODO: AWS validate this assumption
102	# if the RAW_URI starts with double slashes, `urlparse` will fail to decode it as path only
103	# it also means that we already only have the path, so we just need to remove the query string
104	raw_uri = raw_uri.split("?")[0]	2✔
105	raw_path = "/" + raw_uri.lstrip("/")	2✔
106
107	else:
108	# we need to make sure we have a path here, sometimes RAW_URI can be a full URI (when proxied)
109	raw_path = raw_uri = urlparse(raw_uri).path	2✔
110
111	invocation_request["path"] = raw_path	2✔
112	invocation_request["raw_path"] = raw_uri	2✔
113
114	@staticmethod	2✔
115	def _get_single_and_multi_values_from_multidict(	2✔
116	multi_dict: MultiDict,
117	) -> tuple[dict[str, str], dict[str, list[str]]]:
118	single_values = {}	2✔
119	multi_values = defaultdict(list)	2✔
120
121	for key, value in multi_dict.items(multi=True):	2✔
122	multi_values[key].append(value)	2✔
123	# for the single value parameters, AWS only keeps the last value of the list
124	single_values[key] = value	2✔
125
126	return single_values, dict(multi_values)	2✔
127
128	@staticmethod	2✔
129	def _get_invocation_headers(headers: Headers) -> Headers:	2✔
130	invocation_headers = Headers()	2✔
131	for key, value in headers:	2✔
132	if should_drop_header_from_invocation(key):	2✔
133	LOG.debug("Dropping header from invocation request: '%s'", key)	2✔
134	continue	2✔
135	invocation_headers.add(key, value)	2✔
136	return invocation_headers	2✔
137
138	@staticmethod	2✔
139	def create_context_variables(context: RestApiInvocationContext) -> ContextVariables:	2✔
140	invocation_request: InvocationRequest = context.invocation_request	2✔
141	domain_name = invocation_request["headers"].get("Host", "")	2✔
142	domain_prefix = domain_name.split(".")[0]	2✔
143	now = datetime.datetime.now()	2✔
144
145	context_variables = ContextVariables(	2✔
146	accountId=context.account_id,
147	apiId=context.api_id,
148	deploymentId=context.deployment_id,
149	domainName=domain_name,
150	domainPrefix=domain_prefix,
151	extendedRequestId=short_uid(), # TODO: use snapshot tests to verify format
152	httpMethod=invocation_request["http_method"],
153	identity=ContextVarsIdentity(
154	accountId=None,
155	accessKey=None,
156	caller=None,
157	cognitoAuthenticationProvider=None,
158	cognitoAuthenticationType=None,
159	cognitoIdentityId=None,
160	cognitoIdentityPoolId=None,
161	principalOrgId=None,
162	sourceIp="127.0.0.1", # TODO: get the sourceIp from the Request
163	user=None,
164	userAgent=invocation_request["headers"].get("User-Agent"),
165	userArn=None,
166	),
167	path=f"/{context.stage}{invocation_request['raw_path']}",
168	protocol="HTTP/1.1",
169	requestId=long_uid(),
170	requestTime=timestamp(time=now, format=REQUEST_TIME_DATE_FORMAT),
171	requestTimeEpoch=int(now.timestamp() * 1000),
172	stage=context.stage,
173	)
174	if context.is_canary is not None:	2✔
175	context_variables["isCanaryRequest"] = context.is_canary	×
176
177	return context_variables	2✔
178
179	@staticmethod	2✔
180	def get_stage_variables(context: RestApiInvocationContext) -> dict[str, str] \| None:	2✔
181	stage_variables = context.stage_configuration.get("variables")	2✔
182	if context.is_canary:	2✔
183	overrides = (	×
184	context.stage_configuration["canarySettings"].get("stageVariableOverrides") or {}
185	)
186	stage_variables = (stage_variables or {}) \| overrides	×
187
188	if not stage_variables:	2✔
UNCOV 189	return None	1✔
190
191	return stage_variables	2✔
192
193	@staticmethod	2✔
194	def populate_trace_id(headers: Headers) -> str:	2✔
195	incoming_trace = parse_trace_id(headers.get("x-amzn-trace-id", ""))	2✔
196	# parse_trace_id always return capitalized keys
197
198	trace = incoming_trace.get("Root", generate_trace_id())	2✔
199	incoming_parent = incoming_trace.get("Parent")	2✔
200	parent = incoming_parent or generate_trace_parent()	2✔
201	sampled = incoming_trace.get("Sampled", "1" if incoming_parent else "0")	2✔
202	# TODO: lineage? not sure what it related to
203	return f"Root={trace};Parent={parent};Sampled={sampled}"	2✔

localstack / localstack / 20357994067

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