20139e2d-bb2b-4539-9c09-7a23e541ea72

Committed 16 Apr 2025 03:17PM UTC coverage: 86.294% (-0.1%) from 86.419%

Build # 20139e2d-bb2b-4539-9c09-7a23e541ea72

Build Type

push

circleci

Committed by

web-flow

Commit Message

[AWS][Transcribe] Adding fix for validating Audio length (#12450)

Coverage Stats

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

297 existing lines in 11 files now uncovered.

63699 of 73816 relevant lines covered (86.29%)

0.86 hits per line

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

64.29

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

localstack / localstack / 20139e2d-bb2b-4539-9c09-7a23e541ea72

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