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

74.49

/localstack-core/localstack/services/cloudformation/api_utils.py

import logging
import re
from urllib.parse import urlparse

from localstack import config, constants
from localstack.aws.connect import connect_to
from localstack.services.cloudformation.engine.validations import ValidationError
from localstack.services.s3.utils import (
    extract_bucket_name_and_key_from_headers_and_path,
    normalize_bucket_name,
)
from localstack.utils.functions import run_safe
from localstack.utils.http import safe_requests
from localstack.utils.strings import to_str
from localstack.utils.urls import localstack_host

LOG = logging.getLogger(__name__)


def prepare_template_body(req_data: dict) -> str | bytes | None:  # TODO: mutating and returning
    template_url = req_data.get("TemplateURL")
    if template_url:
        req_data["TemplateURL"] = convert_s3_to_local_url(template_url)
    url = req_data.get("TemplateURL", "")
    if is_local_service_url(url):
        modified_template_body = get_template_body(req_data)
        if modified_template_body:
            req_data.pop("TemplateURL", None)
            req_data["TemplateBody"] = modified_template_body
    modified_template_body = get_template_body(req_data)
    if modified_template_body:
        req_data["TemplateBody"] = modified_template_body
    return modified_template_body


def extract_template_body(request: dict) -> str:
    """
    Given a request payload, fetch the body of the template either from S3 or from the payload itself
    """
    if template_body := request.get("TemplateBody"):
        if request.get("TemplateURL"):
            raise ValidationError(
                "Specify exactly one of 'TemplateBody' or 'TemplateUrl'"
            )  # TODO: check proper message

        return template_body

    elif template_url := request.get("TemplateURL"):
        template_url = convert_s3_to_local_url(template_url)
        return get_remote_template_body(template_url)

    else:
        raise ValidationError(
            "Specify exactly one of 'TemplateBody' or 'TemplateUrl'"
        )  # TODO: check proper message


def get_remote_template_body(url: str) -> str:
    response = run_safe(lambda: safe_requests.get(url, verify=False))
    # check error codes, and code 301 - fixes https://github.com/localstack/localstack/issues/1884
    status_code = 0 if response is None else response.status_code
    if 200 <= status_code < 300:
        # request was ok
        return response.text
    elif response is None or status_code == 301 or status_code >= 400:
        # check if this is an S3 URL, then get the file directly from there
        url = convert_s3_to_local_url(url)
        if is_local_service_url(url):
            parsed_path = urlparse(url).path.lstrip("/")
            parts = parsed_path.partition("/")
            client = connect_to().s3
            LOG.debug(
                "Download CloudFormation template content from local S3: %s - %s",
                parts[0],
                parts[2],
            )
            result = client.get_object(Bucket=parts[0], Key=parts[2])
            body = to_str(result["Body"].read())
            return body
        raise RuntimeError(f"Unable to fetch template body (code {status_code}) from URL {url}")
    else:
        raise RuntimeError(
            f"Bad status code from fetching template from url '{url}' ({status_code})",
            url,
            status_code,
        )


def get_template_body(req_data: dict) -> str:
    body = req_data.get("TemplateBody")
    if body:
        return body
    url = req_data.get("TemplateURL")
    if url:
        response = run_safe(lambda: safe_requests.get(url, verify=False))
        # check error codes, and code 301 - fixes https://github.com/localstack/localstack/issues/1884
        status_code = 0 if response is None else response.status_code
        if response is None or status_code == 301 or status_code >= 400:
            # check if this is an S3 URL, then get the file directly from there
            url = convert_s3_to_local_url(url)
            if is_local_service_url(url):
                parsed_path = urlparse(url).path.lstrip("/")
                parts = parsed_path.partition("/")
                client = connect_to().s3
                LOG.debug(
                    "Download CloudFormation template content from local S3: %s - %s",
                    parts[0],
                    parts[2],
                )
                result = client.get_object(Bucket=parts[0], Key=parts[2])
                body = to_str(result["Body"].read())
                return body
            raise Exception(f"Unable to fetch template body (code {status_code}) from URL {url}")
        return to_str(response.content)
    raise Exception(f"Unable to get template body from input: {req_data}")


def is_local_service_url(url: str) -> bool:
    if not url:
        return False
    candidates = (
        constants.LOCALHOST,
        constants.LOCALHOST_HOSTNAME,
        localstack_host().host,
    )
    if any(re.match(rf"^[^:]+://[^:/]*{host}([:/]|$)", url) for host in candidates):
        return True
    host = url.split("://")[-1].split("/")[0]
    return "localhost" in host


def convert_s3_to_local_url(url: str) -> str:
    from localstack.services.cloudformation.provider import ValidationError

    url_parsed = urlparse(url)
    path = url_parsed.path

    headers = {"host": url_parsed.netloc}
    bucket_name, key_name = extract_bucket_name_and_key_from_headers_and_path(headers, path)

    if url_parsed.scheme == "s3":
        raise ValidationError(
            f"S3 error: Domain name specified in {url_parsed.netloc} is not a valid S3 domain"
        )

    if not bucket_name or not key_name:
        if not (url_parsed.netloc.startswith("s3.") or ".s3." in url_parsed.netloc):
            raise ValidationError("TemplateURL must be a supported URL.")

    # note: make sure to normalize the bucket name here!
    bucket_name = normalize_bucket_name(bucket_name)
    local_url = f"{config.internal_service_url()}/{bucket_name}/{key_name}"
    return local_url


def validate_stack_name(stack_name):
    pattern = r"[a-zA-Z][-a-zA-Z0-9]*|arn:[-a-zA-Z0-9:/._+]*"
    return re.match(pattern, stack_name) is not None

1	import logging	2✔
2	import re	2✔
3	from urllib.parse import urlparse	2✔
4
5	from localstack import config, constants	2✔
6	from localstack.aws.connect import connect_to	2✔
7	from localstack.services.cloudformation.engine.validations import ValidationError	2✔
8	from localstack.services.s3.utils import (	2✔
9	extract_bucket_name_and_key_from_headers_and_path,
10	normalize_bucket_name,
11	)
12	from localstack.utils.functions import run_safe	2✔
13	from localstack.utils.http import safe_requests	2✔
14	from localstack.utils.strings import to_str	2✔
15	from localstack.utils.urls import localstack_host	2✔
16
17	LOG = logging.getLogger(__name__)	2✔
18
19
20	def prepare_template_body(req_data: dict) -> str \| bytes \| None: # TODO: mutating and returning	2✔
UNCOV 21	template_url = req_data.get("TemplateURL")	1✔
UNCOV 22	if template_url:	1✔
UNCOV 23	req_data["TemplateURL"] = convert_s3_to_local_url(template_url)	1✔
UNCOV 24	url = req_data.get("TemplateURL", "")	1✔
UNCOV 25	if is_local_service_url(url):	1✔
UNCOV 26	modified_template_body = get_template_body(req_data)	1✔
UNCOV 27	if modified_template_body:	1✔
UNCOV 28	req_data.pop("TemplateURL", None)	1✔
UNCOV 29	req_data["TemplateBody"] = modified_template_body	1✔
UNCOV 30	modified_template_body = get_template_body(req_data)	1✔
UNCOV 31	if modified_template_body:	1✔
UNCOV 32	req_data["TemplateBody"] = modified_template_body	1✔
UNCOV 33	return modified_template_body	1✔
34
35
36	def extract_template_body(request: dict) -> str:	2✔
37	"""
38	Given a request payload, fetch the body of the template either from S3 or from the payload itself
39	"""
40	if template_body := request.get("TemplateBody"):	2✔
41	if request.get("TemplateURL"):	2✔
42	raise ValidationError(	×
43	"Specify exactly one of 'TemplateBody' or 'TemplateUrl'"
44	) # TODO: check proper message
45
46	return template_body	2✔
47
48	elif template_url := request.get("TemplateURL"):	2✔
49	template_url = convert_s3_to_local_url(template_url)	2✔
50	return get_remote_template_body(template_url)	2✔
51
52	else:
53	raise ValidationError(	×
54	"Specify exactly one of 'TemplateBody' or 'TemplateUrl'"
55	) # TODO: check proper message
56
57
58	def get_remote_template_body(url: str) -> str:	2✔
59	response = run_safe(lambda: safe_requests.get(url, verify=False))	2✔
60	# check error codes, and code 301 - fixes https://github.com/localstack/localstack/issues/1884
61	status_code = 0 if response is None else response.status_code	2✔
62	if 200 <= status_code < 300:	2✔
63	# request was ok
64	return response.text	2✔
65	elif response is None or status_code == 301 or status_code >= 400:	×
66	# check if this is an S3 URL, then get the file directly from there
67	url = convert_s3_to_local_url(url)	×
68	if is_local_service_url(url):	×
69	parsed_path = urlparse(url).path.lstrip("/")	×
70	parts = parsed_path.partition("/")	×
71	client = connect_to().s3	×
72	LOG.debug(	×
73	"Download CloudFormation template content from local S3: %s - %s",
74	parts[0],
75	parts[2],
76	)
77	result = client.get_object(Bucket=parts[0], Key=parts[2])	×
78	body = to_str(result["Body"].read())	×
79	return body	×
80	raise RuntimeError(f"Unable to fetch template body (code {status_code}) from URL {url}")	×
81	else:
82	raise RuntimeError(	×
83	f"Bad status code from fetching template from url '{url}' ({status_code})",
84	url,
85	status_code,
86	)
87
88
89	def get_template_body(req_data: dict) -> str:	2✔
90	body = req_data.get("TemplateBody")	2✔
91	if body:	2✔
92	return body	2✔
UNCOV 93	url = req_data.get("TemplateURL")	1✔
UNCOV 94	if url:	1✔
UNCOV 95	response = run_safe(lambda: safe_requests.get(url, verify=False))	1✔
96	# check error codes, and code 301 - fixes https://github.com/localstack/localstack/issues/1884
UNCOV 97	status_code = 0 if response is None else response.status_code	1✔
UNCOV 98	if response is None or status_code == 301 or status_code >= 400:	1✔
99	# check if this is an S3 URL, then get the file directly from there
100	url = convert_s3_to_local_url(url)	×
101	if is_local_service_url(url):	×
102	parsed_path = urlparse(url).path.lstrip("/")	×
103	parts = parsed_path.partition("/")	×
104	client = connect_to().s3	×
105	LOG.debug(	×
106	"Download CloudFormation template content from local S3: %s - %s",
107	parts[0],
108	parts[2],
109	)
110	result = client.get_object(Bucket=parts[0], Key=parts[2])	×
111	body = to_str(result["Body"].read())	×
112	return body	×
113	raise Exception(f"Unable to fetch template body (code {status_code}) from URL {url}")	×
UNCOV 114	return to_str(response.content)	1✔
115	raise Exception(f"Unable to get template body from input: {req_data}")	×
116
117
118	def is_local_service_url(url: str) -> bool:	2✔
119	if not url:	2✔
UNCOV 120	return False	1✔
121	candidates = (	2✔
122	constants.LOCALHOST,
123	constants.LOCALHOST_HOSTNAME,
124	localstack_host().host,
125	)
126	if any(re.match(rf"^[^:]+://[^:/]*{host}([:/]\|$)", url) for host in candidates):	2✔
127	return True	2✔
128	host = url.split("://")[-1].split("/")[0]	2✔
129	return "localhost" in host	2✔
130
131
132	def convert_s3_to_local_url(url: str) -> str:	2✔
133	from localstack.services.cloudformation.provider import ValidationError	2✔
134
135	url_parsed = urlparse(url)	2✔
136	path = url_parsed.path	2✔
137
138	headers = {"host": url_parsed.netloc}	2✔
139	bucket_name, key_name = extract_bucket_name_and_key_from_headers_and_path(headers, path)	2✔
140
141	if url_parsed.scheme == "s3":	2✔
142	raise ValidationError(	2✔
143	f"S3 error: Domain name specified in {url_parsed.netloc} is not a valid S3 domain"
144	)
145
146	if not bucket_name or not key_name:	2✔
147	if not (url_parsed.netloc.startswith("s3.") or ".s3." in url_parsed.netloc):	2✔
148	raise ValidationError("TemplateURL must be a supported URL.")	2✔
149
150	# note: make sure to normalize the bucket name here!
151	bucket_name = normalize_bucket_name(bucket_name)	2✔
152	local_url = f"{config.internal_service_url()}/{bucket_name}/{key_name}"	2✔
153	return local_url	2✔
154
155
156	def validate_stack_name(stack_name):	2✔
UNCOV 157	pattern = r"[a-zA-Z][-a-zA-Z0-9]\|arn:[-a-zA-Z0-9:/._+]"	1✔
UNCOV 158	return re.match(pattern, stack_name) is not None	1✔

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