18505123992

Committed 14 Oct 2025 05:30PM UTC coverage: 86.888% (-0.01%) from 86.899%

Build # 18505123992

Build Type

push

github

Committed by

web-flow

Commit Message

S3: fix `aws-global` validation in CreateBucket (#13250)

Coverage Stats

10 of 10 new or added lines in 4 files covered. (100.0%)

831 existing lines in 40 files now uncovered.

68028 of 78294 relevant lines covered (86.89%)

0.87 hits per line

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

75.47

/localstack-core/localstack/utils/aws/request_context.py

"""
This module has utilities relating to creating/parsing AWS requests.
"""

import logging
import re

from rolo import Request as RoloRequest

from localstack.aws.accounts import get_account_id_from_access_key_id
from localstack.constants import (
    APPLICATION_AMZ_JSON_1_0,
    APPLICATION_AMZ_JSON_1_1,
    APPLICATION_X_WWW_FORM_URLENCODED,
    AWS_REGION_US_EAST_1,
    DEFAULT_AWS_ACCOUNT_ID,
)

LOG = logging.getLogger(__name__)

AWS_REGION_REGEX = r"(us(-gov)?|ap|ca|cn|eu|sa)-(central|(north|south)?(east|west)?)-\d"


def get_account_id_from_request(request: RoloRequest) -> str:
    access_key_id = (
        extract_access_key_id_from_auth_header(request.headers) or DEFAULT_AWS_ACCOUNT_ID
    )

    return get_account_id_from_access_key_id(access_key_id)


def extract_region_from_auth_header(headers) -> str | None:
    auth = headers.get("Authorization") or ""
    region = re.sub(r".*Credential=[^/]+/[^/]+/([^/]+)/.*", r"\1", auth)
    if region == auth:
        return None
    return region


def extract_account_id_from_auth_header(headers) -> str | None:
    if access_key_id := extract_access_key_id_from_auth_header(headers):
        return get_account_id_from_access_key_id(access_key_id)


def extract_access_key_id_from_auth_header(headers: dict[str, str]) -> str | None:
    auth = headers.get("Authorization") or ""

    if auth.startswith("AWS4-"):
        # For Signature Version 4
        access_id = re.findall(r".*Credential=([^/]+)/[^/]+/[^/]+/.*", auth)
        if len(access_id):
            return access_id[0]

    elif auth.startswith("AWS "):
        # For Signature Version 2
        access_id = auth.removeprefix("AWS ").split(":")
        if len(access_id):
            return access_id[0]


def extract_account_id_from_headers(headers) -> str:
    return extract_account_id_from_auth_header(headers) or DEFAULT_AWS_ACCOUNT_ID


def extract_region_from_headers(headers) -> str:
    return extract_region_from_auth_header(headers) or AWS_REGION_US_EAST_1


def extract_service_name_from_auth_header(headers: dict) -> str | None:
    try:
        auth_header = headers.get("authorization", "")
        credential_scope = auth_header.split(",")[0].split()[1]
        _, _, _, service, _ = credential_scope.split("/")
        return service
    except Exception:
        return


def mock_aws_request_headers(
    service: str, aws_access_key_id: str, region_name: str, internal: bool = False
) -> dict[str, str]:
    """
    Returns a mock set of headers that resemble SigV4 signing method.
    """
    from localstack.aws.connect import (
        INTERNAL_REQUEST_PARAMS_HEADER,
        InternalRequestParameters,
        dump_dto,
    )

    ctype = APPLICATION_AMZ_JSON_1_0
    if service == "kinesis":
        ctype = APPLICATION_AMZ_JSON_1_1
    elif service in ["sns", "sqs", "sts", "cloudformation"]:
        ctype = APPLICATION_X_WWW_FORM_URLENCODED

    # For S3 presigned URLs, we require that the client and server use the same
    # access key ID to sign requests. So try to use the access key ID for the
    # current request if available
    headers = {
        "Content-Type": ctype,
        "Accept-Encoding": "identity",
        "X-Amz-Date": "20160623T103251Z",  # TODO: Use current date
        "Authorization": (
            "AWS4-HMAC-SHA256 "
            + f"Credential={aws_access_key_id}/20160623/{region_name}/{service}/aws4_request, "
            + "SignedHeaders=content-type;host;x-amz-date;x-amz-target, Signature=1234"
        ),
    }

    if internal:
        dto = InternalRequestParameters()
        headers[INTERNAL_REQUEST_PARAMS_HEADER] = dump_dto(dto)

    return headers

1	"""
2	This module has utilities relating to creating/parsing AWS requests.
3	"""
4
5	import logging	1✔
6	import re	1✔
7
8	from rolo import Request as RoloRequest	1✔
9
10	from localstack.aws.accounts import get_account_id_from_access_key_id	1✔
11	from localstack.constants import (	1✔
12	APPLICATION_AMZ_JSON_1_0,
13	APPLICATION_AMZ_JSON_1_1,
14	APPLICATION_X_WWW_FORM_URLENCODED,
15	AWS_REGION_US_EAST_1,
16	DEFAULT_AWS_ACCOUNT_ID,
17	)
18
19	LOG = logging.getLogger(__name__)	1✔
20
21	AWS_REGION_REGEX = r"(us(-gov)?\|ap\|ca\|cn\|eu\|sa)-(central\|(north\|south)?(east\|west)?)-\d"	1✔
22
23
24	def get_account_id_from_request(request: RoloRequest) -> str:	1✔
UNCOV 25	access_key_id = (	×
26	extract_access_key_id_from_auth_header(request.headers) or DEFAULT_AWS_ACCOUNT_ID
27	)
28
UNCOV 29	return get_account_id_from_access_key_id(access_key_id)	×
30
31
32	def extract_region_from_auth_header(headers) -> str \| None:	1✔
33	auth = headers.get("Authorization") or ""	1✔
34	region = re.sub(r".Credential=[^/]+/[^/]+/([^/]+)/.", r"\1", auth)	1✔
35	if region == auth:	1✔
36	return None	1✔
37	return region	1✔
38
39
40	def extract_account_id_from_auth_header(headers) -> str \| None:	1✔
UNCOV 41	if access_key_id := extract_access_key_id_from_auth_header(headers):	×
42	return get_account_id_from_access_key_id(access_key_id)	×
43
44
45	def extract_access_key_id_from_auth_header(headers: dict[str, str]) -> str \| None:	1✔
46	auth = headers.get("Authorization") or ""	1✔
47
48	if auth.startswith("AWS4-"):	1✔
49	# For Signature Version 4
50	access_id = re.findall(r".Credential=([^/]+)/[^/]+/[^/]+/.", auth)	1✔
51	if len(access_id):	1✔
52	return access_id[0]	1✔
53
54	elif auth.startswith("AWS "):	1✔
55	# For Signature Version 2
56	access_id = auth.removeprefix("AWS ").split(":")	1✔
57	if len(access_id):	1✔
58	return access_id[0]	1✔
59
60
61	def extract_account_id_from_headers(headers) -> str:	1✔
UNCOV 62	return extract_account_id_from_auth_header(headers) or DEFAULT_AWS_ACCOUNT_ID	×
63
64
65	def extract_region_from_headers(headers) -> str:	1✔
66	return extract_region_from_auth_header(headers) or AWS_REGION_US_EAST_1	1✔
67
68
69	def extract_service_name_from_auth_header(headers: dict) -> str \| None:	1✔
UNCOV 70	try:	×
71	auth_header = headers.get("authorization", "")	×
72	credential_scope = auth_header.split(",")[0].split()[1]	×
73	_, _, _, service, _ = credential_scope.split("/")	×
74	return service	×
75	except Exception:	×
76	return	×
77
78
79	def mock_aws_request_headers(	1✔
80	service: str, aws_access_key_id: str, region_name: str, internal: bool = False
81	) -> dict[str, str]:
82	"""
83	Returns a mock set of headers that resemble SigV4 signing method.
84	"""
85	from localstack.aws.connect import (	1✔
86	INTERNAL_REQUEST_PARAMS_HEADER,
87	InternalRequestParameters,
88	dump_dto,
89	)
90
91	ctype = APPLICATION_AMZ_JSON_1_0	1✔
92	if service == "kinesis":	1✔
UNCOV 93	ctype = APPLICATION_AMZ_JSON_1_1	×
94	elif service in ["sns", "sqs", "sts", "cloudformation"]:	1✔
95	ctype = APPLICATION_X_WWW_FORM_URLENCODED	1✔
96
97	# For S3 presigned URLs, we require that the client and server use the same
98	# access key ID to sign requests. So try to use the access key ID for the
99	# current request if available
100	headers = {	1✔
101	"Content-Type": ctype,
102	"Accept-Encoding": "identity",
103	"X-Amz-Date": "20160623T103251Z", # TODO: Use current date
104	"Authorization": (
105	"AWS4-HMAC-SHA256 "
106	+ f"Credential={aws_access_key_id}/20160623/{region_name}/{service}/aws4_request, "
107	+ "SignedHeaders=content-type;host;x-amz-date;x-amz-target, Signature=1234"
108	),
109	}
110
111	if internal:	1✔
112	dto = InternalRequestParameters()	1✔
113	headers[INTERNAL_REQUEST_PARAMS_HEADER] = dump_dto(dto)	1✔
114
115	return headers	1✔

localstack / localstack / 18505123992

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