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

94.83

/localstack-core/localstack/services/lambda_/urlrouter.py

"""Routing for Lambda function URLs: https://docs.aws.amazon.com/lambda/latest/dg/lambda-urls.html"""

import base64
import json
import logging
from datetime import datetime
from http import HTTPStatus
from json import JSONDecodeError

from rolo.request import restore_payload

from localstack.aws.api.lambda_ import InvocationType
from localstack.aws.protocol.serializer import gen_amzn_requestid
from localstack.http import Request, Response, Router
from localstack.http.dispatcher import Handler
from localstack.services.lambda_.api_utils import FULL_FN_ARN_PATTERN
from localstack.services.lambda_.invocation.lambda_models import InvocationResult
from localstack.services.lambda_.invocation.lambda_service import LambdaService
from localstack.services.lambda_.invocation.models import lambda_stores
from localstack.utils.aws.request_context import AWS_REGION_REGEX
from localstack.utils.strings import long_uid, to_bytes, to_str
from localstack.utils.time import TIMESTAMP_READABLE_FORMAT, mktime, timestamp
from localstack.utils.urls import localstack_host

LOG = logging.getLogger(__name__)


class FunctionUrlRouter:
    router: Router[Handler]
    lambda_service: LambdaService

    def __init__(self, router: Router[Handler], lambda_service: LambdaService):
        self.router = router
        self.registered = False
        self.lambda_service = lambda_service

    def register_routes(self) -> None:
        if self.registered:
            LOG.debug("Skipped Lambda URL route registration (routes already registered).")
            return
        self.registered = True

        LOG.debug("Registering parameterized Lambda routes.")

        self.router.add(
            "/",
            host=f"<api_id>.lambda-url.<regex('{AWS_REGION_REGEX}'):region>.<regex('.*'):server>",
            endpoint=self.handle_lambda_url_invocation,
            defaults={"path": ""},
        )
        self.router.add(
            "/<path:path>",
            host=f"<api_id>.lambda-url.<regex('{AWS_REGION_REGEX}'):region>.<regex('.*'):server>",
            endpoint=self.handle_lambda_url_invocation,
        )

    def handle_lambda_url_invocation(
        self,
        request: Request,
        api_id: str,
        region: str,
        **url_params: str,
    ) -> Response:
        response = Response()
        response.mimetype = "application/json"

        lambda_url_config = None

        for account_id in lambda_stores.keys():
            store = lambda_stores[account_id][region]
            for fn in store.functions.values():
                for url_config in fn.function_url_configs.values():
                    # AWS tags are case sensitive, but domains are not.
                    # So we normalize them here to maximize both AWS and RFC
                    # conformance
                    if url_config.url_id.lower() == api_id.lower():
                        lambda_url_config = url_config

        # TODO: check if errors are different when the URL has existed previously
        if lambda_url_config is None:
            LOG.info("Lambda URL %s does not exist", request.url)
            response.data = '{"Message":null}'
            response.status = 403
            response.headers["x-amzn-ErrorType"] = "AccessDeniedException"
            # TODO: x-amzn-requestid
            return response

        event = event_for_lambda_url(api_id, request)

        match = FULL_FN_ARN_PATTERN.search(lambda_url_config.function_arn).groupdict()

        result = self.lambda_service.invoke(
            function_name=match.get("function_name"),
            qualifier=match.get("qualifier"),
            account_id=match.get("account_id"),
            region=match.get("region_name"),
            invocation_type=InvocationType.RequestResponse,
            client_context="{}",  # TODO: test
            payload=to_bytes(json.dumps(event)),
            request_id=gen_amzn_requestid(),
        )
        if result.is_error:
            response = Response("Internal Server Error", HTTPStatus.BAD_GATEWAY)
        else:
            response = lambda_result_to_response(result)
        return response


def event_for_lambda_url(api_id: str, request: Request) -> dict:
    partitioned_uri = request.full_path.partition("?")
    raw_path = partitioned_uri[0]
    raw_query_string = partitioned_uri[2]

    query_string_parameters = {k: ",".join(request.args.getlist(k)) for k in request.args.keys()}

    now = datetime.utcnow()
    readable = timestamp(time=now, format=TIMESTAMP_READABLE_FORMAT)
    if not any(char in readable for char in ["+", "-"]):
        readable += "+0000"

    data = restore_payload(request)
    headers = request.headers
    source_ip = headers.get("Remote-Addr", "")
    request_context = {
        "accountId": "anonymous",
        "apiId": api_id,
        "domainName": headers.get("Host", ""),
        "domainPrefix": api_id,
        "http": {
            "method": request.method,
            "path": raw_path,
            "protocol": "HTTP/1.1",
            "sourceIp": source_ip,
            "userAgent": headers.get("User-Agent", ""),
        },
        "requestId": long_uid(),
        "routeKey": "$default",
        "stage": "$default",
        "time": readable,
        "timeEpoch": mktime(ts=now, millis=True),
    }

    content_type = headers.get("Content-Type", "").lower()
    content_type_is_text = any(text_type in content_type for text_type in ["text", "json", "xml"])

    is_base64_encoded = not (data.isascii() and content_type_is_text) if data else False
    body = base64.b64encode(data).decode() if is_base64_encoded else data
    if isinstance(body, bytes):
        body = to_str(body)

    ignored_headers = ["connection", "x-localstack-tgt-api", "x-localstack-request-url"]
    event_headers = {k.lower(): v for k, v in headers.items() if k.lower() not in ignored_headers}

    event_headers.update(
        {
            "x-amzn-tls-cipher-suite": "ECDHE-RSA-AES128-GCM-SHA256",
            "x-amzn-tls-version": "TLSv1.2",
            "x-forwarded-proto": "http",
            "x-forwarded-for": source_ip,
            "x-forwarded-port": str(localstack_host().port),
        }
    )

    event = {
        "version": "2.0",
        "routeKey": "$default",
        "rawPath": raw_path,
        "rawQueryString": raw_query_string,
        "headers": event_headers,
        "queryStringParameters": query_string_parameters,
        "requestContext": request_context,
        "body": body,
        "isBase64Encoded": is_base64_encoded,
    }

    if not data:
        event.pop("body")

    return event


def lambda_result_to_response(result: InvocationResult):
    response = Response()

    # Set default headers
    response.headers.update(
        {
            "Content-Type": "application/json",
            "Connection": "keep-alive",
            "x-amzn-requestid": result.request_id,
            "x-amzn-trace-id": long_uid(),  # TODO: get the proper trace id here
        }
    )

    original_payload = to_str(result.payload)
    try:
        parsed_result = json.loads(original_payload)
    except JSONDecodeError:
        # URL router must be able to parse a Streaming Response without necessary defining it in the URL Config
        # And if the body is a simple string, it should be returned without issues
        split_index = original_payload.find("\x00" * 8)
        if split_index == -1:
            parsed_result = {"body": original_payload}
        else:
            metadata = original_payload[:split_index]
            body_str = original_payload[split_index + 8 :]
            parsed_result = {**json.loads(metadata), "body": body_str}

    # patch to fix whitespaces
    # TODO: check if this is a downstream issue of invocation result serialization
    original_payload = json.dumps(parsed_result, separators=(",", ":"))

    if isinstance(parsed_result, str):
        # a string is a special case here and is returned as-is
        response.data = parsed_result

    elif isinstance(parsed_result, dict):
        # if it's a dict it might be a proper response
        if isinstance(parsed_result.get("headers"), dict):
            response.headers.update(parsed_result.get("headers"))
        if "statusCode" in parsed_result:
            response.status_code = int(parsed_result["statusCode"])
        if "body" not in parsed_result:
            # TODO: test if providing a status code but no body actually works
            response.data = original_payload
        elif isinstance(parsed_result.get("body"), dict):
            response.data = json.dumps(parsed_result.get("body"))
        elif parsed_result.get("isBase64Encoded", False):
            body_bytes = to_bytes(to_str(parsed_result.get("body", "")))
            decoded_body_bytes = base64.b64decode(body_bytes)
            response.data = decoded_body_bytes
        else:
            response.data = parsed_result.get("body")
    else:
        response.data = original_payload

    return response

1	"""Routing for Lambda function URLs: https://docs.aws.amazon.com/lambda/latest/dg/lambda-urls.html"""
2
3	import base64	1✔
4	import json	1✔
5	import logging	1✔
6	from datetime import datetime	1✔
7	from http import HTTPStatus	1✔
8	from json import JSONDecodeError	1✔
9
10	from rolo.request import restore_payload	1✔
11
12	from localstack.aws.api.lambda_ import InvocationType	1✔
13	from localstack.aws.protocol.serializer import gen_amzn_requestid	1✔
14	from localstack.http import Request, Response, Router	1✔
15	from localstack.http.dispatcher import Handler	1✔
16	from localstack.services.lambda_.api_utils import FULL_FN_ARN_PATTERN	1✔
17	from localstack.services.lambda_.invocation.lambda_models import InvocationResult	1✔
18	from localstack.services.lambda_.invocation.lambda_service import LambdaService	1✔
19	from localstack.services.lambda_.invocation.models import lambda_stores	1✔
20	from localstack.utils.aws.request_context import AWS_REGION_REGEX	1✔
21	from localstack.utils.strings import long_uid, to_bytes, to_str	1✔
22	from localstack.utils.time import TIMESTAMP_READABLE_FORMAT, mktime, timestamp	1✔
23	from localstack.utils.urls import localstack_host	1✔
24
25	LOG = logging.getLogger(__name__)	1✔
26
27
28	class FunctionUrlRouter:	1✔
29	router: Router[Handler]	1✔
30	lambda_service: LambdaService	1✔
31
32	def __init__(self, router: Router[Handler], lambda_service: LambdaService):	1✔
33	self.router = router	1✔
34	self.registered = False	1✔
35	self.lambda_service = lambda_service	1✔
36
37	def register_routes(self) -> None:	1✔
38	if self.registered:	1✔
39	LOG.debug("Skipped Lambda URL route registration (routes already registered).")	×
UNCOV 40	return	×
41	self.registered = True	1✔
42
43	LOG.debug("Registering parameterized Lambda routes.")	1✔
44
45	self.router.add(	1✔
46	"/",
47	host=f"<api_id>.lambda-url.<regex('{AWS_REGION_REGEX}'):region>.<regex('.*'):server>",
48	endpoint=self.handle_lambda_url_invocation,
49	defaults={"path": ""},
50	)
51	self.router.add(	1✔
52	"/<path:path>",
53	host=f"<api_id>.lambda-url.<regex('{AWS_REGION_REGEX}'):region>.<regex('.*'):server>",
54	endpoint=self.handle_lambda_url_invocation,
55	)
56
57	def handle_lambda_url_invocation(	1✔
58	self,
59	request: Request,
60	api_id: str,
61	region: str,
62	**url_params: str,
63	) -> Response:
64	response = Response()	1✔
65	response.mimetype = "application/json"	1✔
66
67	lambda_url_config = None	1✔
68
69	for account_id in lambda_stores.keys():	1✔
70	store = lambda_stores[account_id][region]	1✔
71	for fn in store.functions.values():	1✔
72	for url_config in fn.function_url_configs.values():	1✔
73	# AWS tags are case sensitive, but domains are not.
74	# So we normalize them here to maximize both AWS and RFC
75	# conformance
76	if url_config.url_id.lower() == api_id.lower():	1✔
77	lambda_url_config = url_config	1✔
78
79	# TODO: check if errors are different when the URL has existed previously
80	if lambda_url_config is None:	1✔
81	LOG.info("Lambda URL %s does not exist", request.url)	1✔
82	response.data = '{"Message":null}'	1✔
83	response.status = 403	1✔
84	response.headers["x-amzn-ErrorType"] = "AccessDeniedException"	1✔
85	# TODO: x-amzn-requestid
86	return response	1✔
87
88	event = event_for_lambda_url(api_id, request)	1✔
89
90	match = FULL_FN_ARN_PATTERN.search(lambda_url_config.function_arn).groupdict()	1✔
91
92	result = self.lambda_service.invoke(	1✔
93	function_name=match.get("function_name"),
94	qualifier=match.get("qualifier"),
95	account_id=match.get("account_id"),
96	region=match.get("region_name"),
97	invocation_type=InvocationType.RequestResponse,
98	client_context="{}", # TODO: test
99	payload=to_bytes(json.dumps(event)),
100	request_id=gen_amzn_requestid(),
101	)
102	if result.is_error:	1✔
103	response = Response("Internal Server Error", HTTPStatus.BAD_GATEWAY)	1✔
104	else:
105	response = lambda_result_to_response(result)	1✔
106	return response	1✔
107
108
109	def event_for_lambda_url(api_id: str, request: Request) -> dict:	1✔
110	partitioned_uri = request.full_path.partition("?")	1✔
111	raw_path = partitioned_uri[0]	1✔
112	raw_query_string = partitioned_uri[2]	1✔
113
114	query_string_parameters = {k: ",".join(request.args.getlist(k)) for k in request.args.keys()}	1✔
115
116	now = datetime.utcnow()	1✔
117	readable = timestamp(time=now, format=TIMESTAMP_READABLE_FORMAT)	1✔
118	if not any(char in readable for char in ["+", "-"]):	1✔
119	readable += "+0000"	1✔
120
121	data = restore_payload(request)	1✔
122	headers = request.headers	1✔
123	source_ip = headers.get("Remote-Addr", "")	1✔
124	request_context = {	1✔
125	"accountId": "anonymous",
126	"apiId": api_id,
127	"domainName": headers.get("Host", ""),
128	"domainPrefix": api_id,
129	"http": {
130	"method": request.method,
131	"path": raw_path,
132	"protocol": "HTTP/1.1",
133	"sourceIp": source_ip,
134	"userAgent": headers.get("User-Agent", ""),
135	},
136	"requestId": long_uid(),
137	"routeKey": "$default",
138	"stage": "$default",
139	"time": readable,
140	"timeEpoch": mktime(ts=now, millis=True),
141	}
142
143	content_type = headers.get("Content-Type", "").lower()	1✔
144	content_type_is_text = any(text_type in content_type for text_type in ["text", "json", "xml"])	1✔
145
146	is_base64_encoded = not (data.isascii() and content_type_is_text) if data else False	1✔
147	body = base64.b64encode(data).decode() if is_base64_encoded else data	1✔
148	if isinstance(body, bytes):	1✔
149	body = to_str(body)	1✔
150
151	ignored_headers = ["connection", "x-localstack-tgt-api", "x-localstack-request-url"]	1✔
152	event_headers = {k.lower(): v for k, v in headers.items() if k.lower() not in ignored_headers}	1✔
153
154	event_headers.update(	1✔
155	{
156	"x-amzn-tls-cipher-suite": "ECDHE-RSA-AES128-GCM-SHA256",
157	"x-amzn-tls-version": "TLSv1.2",
158	"x-forwarded-proto": "http",
159	"x-forwarded-for": source_ip,
160	"x-forwarded-port": str(localstack_host().port),
161	}
162	)
163
164	event = {	1✔
165	"version": "2.0",
166	"routeKey": "$default",
167	"rawPath": raw_path,
168	"rawQueryString": raw_query_string,
169	"headers": event_headers,
170	"queryStringParameters": query_string_parameters,
171	"requestContext": request_context,
172	"body": body,
173	"isBase64Encoded": is_base64_encoded,
174	}
175
176	if not data:	1✔
177	event.pop("body")	1✔
178
179	return event	1✔
180
181
182	def lambda_result_to_response(result: InvocationResult):	1✔
183	response = Response()	1✔
184
185	# Set default headers
186	response.headers.update(	1✔
187	{
188	"Content-Type": "application/json",
189	"Connection": "keep-alive",
190	"x-amzn-requestid": result.request_id,
191	"x-amzn-trace-id": long_uid(), # TODO: get the proper trace id here
192	}
193	)
194
195	original_payload = to_str(result.payload)	1✔
196	try:	1✔
197	parsed_result = json.loads(original_payload)	1✔
198	except JSONDecodeError:	1✔
199	# URL router must be able to parse a Streaming Response without necessary defining it in the URL Config
200	# And if the body is a simple string, it should be returned without issues
201	split_index = original_payload.find("\x00" * 8)	1✔
202	if split_index == -1:	1✔
UNCOV 203	parsed_result = {"body": original_payload}	×
204	else:
205	metadata = original_payload[:split_index]	1✔
206	body_str = original_payload[split_index + 8 :]	1✔
207	parsed_result = {**json.loads(metadata), "body": body_str}	1✔
208
209	# patch to fix whitespaces
210	# TODO: check if this is a downstream issue of invocation result serialization
211	original_payload = json.dumps(parsed_result, separators=(",", ":"))	1✔
212
213	if isinstance(parsed_result, str):	1✔
214	# a string is a special case here and is returned as-is
215	response.data = parsed_result	1✔
216
217	elif isinstance(parsed_result, dict):	1✔
218	# if it's a dict it might be a proper response
219	if isinstance(parsed_result.get("headers"), dict):	1✔
220	response.headers.update(parsed_result.get("headers"))	1✔
221	if "statusCode" in parsed_result:	1✔
222	response.status_code = int(parsed_result["statusCode"])	1✔
223	if "body" not in parsed_result:	1✔
224	# TODO: test if providing a status code but no body actually works
225	response.data = original_payload	1✔
226	elif isinstance(parsed_result.get("body"), dict):	1✔
227	response.data = json.dumps(parsed_result.get("body"))	1✔
228	elif parsed_result.get("isBase64Encoded", False):	1✔
UNCOV 229	body_bytes = to_bytes(to_str(parsed_result.get("body", "")))	×
UNCOV 230	decoded_body_bytes = base64.b64decode(body_bytes)	×
UNCOV 231	response.data = decoded_body_bytes	×
232	else:
233	response.data = parsed_result.get("body")	1✔
234	else:
235	response.data = original_payload	1✔
236
237	return response	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