17481980735

Committed 04 Sep 2025 08:05PM UTC coverage: 86.84% (-0.03%) from 86.867%

Build # 17481980735

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: fix community issues after attempted pro fixes (#13101)

Coverage Stats

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

543 existing lines in 23 files now uncovered.

67097 of 77265 relevant lines covered (86.84%)

0.87 hits per line

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

91.58

/localstack-core/localstack/services/s3/validation.py

import base64
import datetime
import hashlib
from zoneinfo import ZoneInfo

from botocore.utils import InvalidArnException

from localstack.aws.api import CommonServiceException
from localstack.aws.api.s3 import (
    AccessControlPolicy,
    AnalyticsConfiguration,
    AnalyticsId,
    BucketCannedACL,
    BucketLifecycleConfiguration,
    BucketName,
    ChecksumAlgorithm,
    CORSConfiguration,
    Grant,
    Grantee,
    Grants,
    IntelligentTieringConfiguration,
    IntelligentTieringId,
    InvalidArgument,
    InvalidBucketName,
    InvalidEncryptionAlgorithmError,
    InventoryConfiguration,
    InventoryId,
    KeyTooLongError,
    ObjectCannedACL,
    Permission,
    ServerSideEncryption,
    SSECustomerAlgorithm,
    SSECustomerKey,
    SSECustomerKeyMD5,
    WebsiteConfiguration,
)
from localstack.aws.api.s3 import Type as GranteeType
from localstack.services.s3 import constants as s3_constants
from localstack.services.s3.exceptions import InvalidRequest, MalformedACLError, MalformedXML
from localstack.services.s3.utils import (
    get_class_attrs_from_spec_class,
    get_permission_header_name,
    is_bucket_name_valid,
    is_valid_canonical_id,
    validate_dict_fields,
)
from localstack.utils.aws import arns
from localstack.utils.strings import to_bytes

# https://docs.aws.amazon.com/AmazonS3/latest/userguide/acl-overview.html#canned-acl
# bucket-owner-read + bucket-owner-full-control are allowed, but ignored for buckets
VALID_CANNED_ACLS = get_class_attrs_from_spec_class(
    BucketCannedACL
) | get_class_attrs_from_spec_class(ObjectCannedACL)


def validate_bucket_analytics_configuration(
    id: AnalyticsId, analytics_configuration: AnalyticsConfiguration
) -> None:
    if id != analytics_configuration.get("Id"):
        raise MalformedXML(
            "The XML you provided was not well-formed or did not validate against our published schema"
        )


def validate_bucket_intelligent_tiering_configuration(
    id: IntelligentTieringId, intelligent_tiering_configuration: IntelligentTieringConfiguration
) -> None:
    if id != intelligent_tiering_configuration.get("Id"):
        raise MalformedXML(
            "The XML you provided was not well-formed or did not validate against our published schema"
        )


def validate_bucket_name(bucket: BucketName) -> None:
    """
    Validate s3 bucket name based on the documentation
    ref. https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
    """
    if not is_bucket_name_valid(bucket_name=bucket):
        raise InvalidBucketName("The specified bucket is not valid.", BucketName=bucket)


def validate_canned_acl(canned_acl: str) -> None:
    """
    Validate the canned ACL value, or raise an Exception
    """
    if canned_acl and canned_acl not in VALID_CANNED_ACLS:
        raise InvalidArgument(
            None,
            ArgumentName="x-amz-acl",
            ArgumentValue=canned_acl,
        )


def parse_grants_in_headers(permission: Permission, grantees: str) -> Grants:
    splitted_grantees = [grantee.strip() for grantee in grantees.split(",")]
    grants = []
    for seralized_grantee in splitted_grantees:
        grantee_type, grantee_id = seralized_grantee.split("=")
        grantee_id = grantee_id.strip('"')
        if grantee_type not in ("uri", "id", "emailAddress"):
            raise InvalidArgument(
                "Argument format not recognized",
                ArgumentName=get_permission_header_name(permission),
                ArgumentValue=seralized_grantee,
            )
        elif grantee_type == "uri":
            if grantee_id not in s3_constants.VALID_ACL_PREDEFINED_GROUPS:
                raise InvalidArgument(
                    "Invalid group uri",
                    ArgumentName="uri",
                    ArgumentValue=grantee_id,
                )
            grantee = Grantee(
                Type=GranteeType.Group,
                URI=grantee_id,
            )

        elif grantee_type == "id":
            if not is_valid_canonical_id(grantee_id):
                raise InvalidArgument(
                    "Invalid id",
                    ArgumentName="id",
                    ArgumentValue=grantee_id,
                )
            grantee = Grantee(
                Type=GranteeType.CanonicalUser,
                ID=grantee_id,
                DisplayName="webfile",  # TODO: only in certain regions
            )

        else:
            # TODO: check validation here
            grantee = Grantee(
                Type=GranteeType.AmazonCustomerByEmail,
                EmailAddress=grantee_id,
            )
        grants.append(Grant(Permission=permission, Grantee=grantee))

    return grants


def validate_acl_acp(acp: AccessControlPolicy) -> None:
    if acp is None or "Owner" not in acp or "Grants" not in acp:
        raise MalformedACLError(
            "The XML you provided was not well-formed or did not validate against our published schema"
        )

    if not is_valid_canonical_id(owner_id := acp["Owner"].get("ID", "")):
        raise InvalidArgument(
            "Invalid id",
            ArgumentName="CanonicalUser/ID",
            ArgumentValue=owner_id,
        )

    for grant in acp["Grants"]:
        if grant.get("Permission") not in s3_constants.VALID_GRANTEE_PERMISSIONS:
            raise MalformedACLError(
                "The XML you provided was not well-formed or did not validate against our published schema"
            )

        grantee = grant.get("Grantee", {})
        grant_type = grantee.get("Type")
        if grant_type not in (
            GranteeType.Group,
            GranteeType.CanonicalUser,
            GranteeType.AmazonCustomerByEmail,
        ):
            raise MalformedACLError(
                "The XML you provided was not well-formed or did not validate against our published schema"
            )
        elif (
            grant_type == GranteeType.Group
            and (grant_uri := grantee.get("URI", ""))
            not in s3_constants.VALID_ACL_PREDEFINED_GROUPS
        ):
            raise InvalidArgument(
                "Invalid group uri",
                ArgumentName="Group/URI",
                ArgumentValue=grant_uri,
            )

        elif grant_type == GranteeType.AmazonCustomerByEmail:
            # TODO: add validation here
            continue

        elif grant_type == GranteeType.CanonicalUser and not is_valid_canonical_id(
            grantee_id := grantee.get("ID", "")
        ):
            raise InvalidArgument(
                "Invalid id",
                ArgumentName="CanonicalUser/ID",
                ArgumentValue=grantee_id,
            )


def validate_lifecycle_configuration(lifecycle_conf: BucketLifecycleConfiguration) -> None:
    """
    Validate the Lifecycle configuration following AWS docs
    See https://docs.aws.amazon.com/AmazonS3/latest/userguide/intro-lifecycle-rules.html
    https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketLifecycleConfiguration.html
    :param lifecycle_conf: the bucket lifecycle configuration given by the client
    :raises MalformedXML: when the file doesn't follow the basic structure/required fields
    :raises InvalidArgument: if the `Date` passed for the Expiration is not at Midnight GMT
    :raises InvalidRequest: if there are duplicate tags keys in `Tags` field
    :return: None
    """
    # we only add the `Expiration` header, we don't delete objects yet
    # We don't really expire or transition objects
    # TODO: transition not supported not validated, as we don't use it yet
    if not lifecycle_conf:
        return

    for rule in lifecycle_conf.get("Rules", []):
        if any(req_key not in rule for req_key in ("ID", "Filter", "Status")):
            raise MalformedXML()
        if (non_current_exp := rule.get("NoncurrentVersionExpiration")) is not None:
            if all(
                req_key not in non_current_exp
                for req_key in ("NewerNoncurrentVersions", "NoncurrentDays")
            ):
                raise MalformedXML()

        if rule_filter := rule.get("Filter"):
            if len(rule_filter) > 1:
                raise MalformedXML()

        if (expiration := rule.get("Expiration", {})) and "ExpiredObjectDeleteMarker" in expiration:
            if len(expiration) > 1:
                raise MalformedXML()

        if exp_date := (expiration.get("Date")):
            if exp_date.timetz() != datetime.time(
                hour=0, minute=0, second=0, microsecond=0, tzinfo=ZoneInfo("GMT")
            ):
                raise InvalidArgument(
                    "'Date' must be at midnight GMT",
                    ArgumentName="Date",
                    ArgumentValue=exp_date.astimezone(),  # use the locale timezone, that's what AWS does (returns PST?)
                )

        if tags := (rule_filter.get("And", {}).get("Tags")):
            tag_keys = set()
            for tag in tags:
                if (tag_key := tag.get("Key")) in tag_keys:
                    raise InvalidRequest("Duplicate Tag Keys are not allowed.")
                tag_keys.add(tag_key)


def validate_website_configuration(website_config: WebsiteConfiguration) -> None:
    """
    Validate the website configuration following AWS docs
    See https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketWebsite.html
    :param website_config:
    :raises
    :return: None
    """
    if redirect_all_req := website_config.get("RedirectAllRequestsTo", {}):
        if len(website_config) > 1:
            raise InvalidArgument(
                "RedirectAllRequestsTo cannot be provided in conjunction with other Routing Rules.",
                ArgumentName="RedirectAllRequestsTo",
                ArgumentValue="not null",
            )

        if "HostName" not in redirect_all_req:
            raise MalformedXML()

        if (protocol := redirect_all_req.get("Protocol")) and protocol not in ("http", "https"):
            raise InvalidRequest(
                "Invalid protocol, protocol can be http or https. If not defined the protocol will be selected automatically."
            )

        return

    # required
    # https://docs.aws.amazon.com/AmazonS3/latest/API/API_IndexDocument.html
    if not (index_configuration := website_config.get("IndexDocument")):
        raise InvalidArgument(
            "A value for IndexDocument Suffix must be provided if RedirectAllRequestsTo is empty",
            ArgumentName="IndexDocument",
            ArgumentValue="null",
        )

    if not (index_suffix := index_configuration.get("Suffix")) or "/" in index_suffix:
        raise InvalidArgument(
            "The IndexDocument Suffix is not well formed",
            ArgumentName="IndexDocument",
            ArgumentValue=index_suffix or None,
        )

    if "ErrorDocument" in website_config and not website_config.get("ErrorDocument", {}).get("Key"):
        raise MalformedXML()

    if "RoutingRules" in website_config:
        routing_rules = website_config.get("RoutingRules", [])
        if len(routing_rules) == 0:
            raise MalformedXML()
        if len(routing_rules) > 50:
            raise ValueError("Too many routing rules")  # TODO: correct exception
        for routing_rule in routing_rules:
            redirect = routing_rule.get("Redirect", {})
            # todo: this does not raise an error? check what GetWebsiteConfig returns? empty field?
            # if not (redirect := routing_rule.get("Redirect")):
            #     raise "Something"

            if "ReplaceKeyPrefixWith" in redirect and "ReplaceKeyWith" in redirect:
                raise InvalidRequest(
                    "You can only define ReplaceKeyPrefix or ReplaceKey but not both."
                )

            if "Condition" in routing_rule and not routing_rule.get("Condition", {}):
                raise InvalidRequest(
                    "Condition cannot be empty. To redirect all requests without a condition, the condition element shouldn't be present."
                )

            if (protocol := redirect.get("Protocol")) and protocol not in ("http", "https"):
                raise InvalidRequest(
                    "Invalid protocol, protocol can be http or https. If not defined the protocol will be selected automatically."
                )


def validate_inventory_configuration(
    config_id: InventoryId, inventory_configuration: InventoryConfiguration
):
    """
    Validate the Inventory Configuration following AWS docs
    Validation order is XML then `Id` then S3DestinationBucket
    https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketInventoryConfiguration.html
    https://docs.aws.amazon.com/AmazonS3/latest/userguide/storage-inventory.html
    :param config_id: the passed Id parameter passed to the provider method
    :param inventory_configuration: InventoryConfiguration
    :raises MalformedXML: when the file doesn't follow the basic structure/required fields
    :raises IdMismatch: if the `Id` parameter is different from the `Id` field from the configuration
    :raises InvalidS3DestinationBucket: if S3 bucket is not provided as an ARN
    :return: None
    """
    required_root_fields = {"Destination", "Id", "IncludedObjectVersions", "IsEnabled", "Schedule"}
    optional_root_fields = {"Filter", "OptionalFields"}

    if not validate_dict_fields(
        inventory_configuration, required_root_fields, optional_root_fields
    ):
        raise MalformedXML()

    required_s3_bucket_dest_fields = {"Bucket", "Format"}
    optional_s3_bucket_dest_fields = {"AccountId", "Encryption", "Prefix"}

    if not (
        s3_bucket_destination := inventory_configuration["Destination"].get("S3BucketDestination")
    ) or not validate_dict_fields(
        s3_bucket_destination, required_s3_bucket_dest_fields, optional_s3_bucket_dest_fields
    ):
        raise MalformedXML()

    if inventory_configuration["Destination"]["S3BucketDestination"]["Format"] not in (
        "CSV",
        "ORC",
        "Parquet",
    ):
        raise MalformedXML()

    if not (frequency := inventory_configuration["Schedule"].get("Frequency")) or frequency not in (
        "Daily",
        "Weekly",
    ):
        raise MalformedXML()

    if inventory_configuration["IncludedObjectVersions"] not in ("All", "Current"):
        raise MalformedXML()

    possible_optional_fields = {
        "Size",
        "LastModifiedDate",
        "StorageClass",
        "ETag",
        "IsMultipartUploaded",
        "ReplicationStatus",
        "EncryptionStatus",
        "ObjectLockRetainUntilDate",
        "ObjectLockMode",
        "ObjectLockLegalHoldStatus",
        "IntelligentTieringAccessTier",
        "BucketKeyStatus",
        "ChecksumAlgorithm",
    }
    if (opt_fields := inventory_configuration.get("OptionalFields")) and set(
        opt_fields
    ) - possible_optional_fields:
        raise MalformedXML()

    if inventory_configuration.get("Id") != config_id:
        raise CommonServiceException(
            code="IdMismatch", message="Document ID does not match the specified configuration ID."
        )

    bucket_arn = inventory_configuration["Destination"]["S3BucketDestination"]["Bucket"]
    try:
        arns.parse_arn(bucket_arn)
    except InvalidArnException:
        raise CommonServiceException(
            code="InvalidS3DestinationBucket", message="Invalid bucket ARN."
        )


def validate_cors_configuration(cors_configuration: CORSConfiguration):
    rules = cors_configuration["CORSRules"]

    if not rules or len(rules) > 100:
        raise MalformedXML()

    required_rule_fields = {"AllowedMethods", "AllowedOrigins"}
    optional_rule_fields = {"AllowedHeaders", "ExposeHeaders", "MaxAgeSeconds", "ID"}

    for rule in rules:
        if not validate_dict_fields(rule, required_rule_fields, optional_rule_fields):
            raise MalformedXML()

        for method in rule["AllowedMethods"]:
            if method not in ("GET", "PUT", "HEAD", "POST", "DELETE"):
                raise InvalidRequest(
                    f"Found unsupported HTTP method in CORS config. Unsupported method is {method}"
                )


def validate_object_key(object_key: str) -> None:
    """
    ref. https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-keys.html

    """
    if (len_key := len(to_bytes(object_key, encoding="UTF-8"))) > 1024:
        raise KeyTooLongError(
            "Your key is too long",
            MaxSizeAllowed="1024",
            Size=str(len_key),
        )


def validate_sse_c(
    algorithm: SSECustomerAlgorithm,
    encryption_key: SSECustomerKey,
    encryption_key_md5: SSECustomerKeyMD5,
    server_side_encryption: ServerSideEncryption = None,
):
    """
    This method validates the SSE Customer parameters for different requests.
    :param algorithm: the SSECustomerAlgorithm parameter of the incoming Request, can only be AES256
    :param encryption_key: the SSECustomerKey of the incoming Request, represent the base64 encoded encryption key
    :param encryption_key_md5: the SSECustomerKeyMD5 of the request, represents the base64 encoded MD5 hash of the
    encryption key
    :param server_side_encryption: when the incoming request is a "write" request (PutObject, CopyObject,
     CreateMultipartUpload), the user can specify the encryption. Customer encryption and AWS SSE can't both be set.
    :raises: InvalidArgument if the request is invalid
    :raises: InvalidEncryptionAlgorithmError if the given algorithm is different from AES256
    """
    if not encryption_key and not algorithm:
        return
    elif server_side_encryption:
        raise InvalidArgument(
            "Server Side Encryption with Customer provided key is incompatible with the encryption method specified",
            ArgumentName="x-amz-server-side-encryption",
            ArgumentValue=server_side_encryption,
        )

    if encryption_key and not algorithm:
        raise InvalidArgument(
            "Requests specifying Server Side Encryption with Customer provided keys must provide a valid encryption algorithm.",
            ArgumentName="x-amz-server-side-encryption",
            ArgumentValue="null",
        )
    elif not encryption_key and algorithm:
        raise InvalidArgument(
            "Requests specifying Server Side Encryption with Customer provided keys must provide an appropriate secret key.",
            ArgumentName="x-amz-server-side-encryption",
            ArgumentValue="null",
        )

    if algorithm != "AES256":
        raise InvalidEncryptionAlgorithmError(
            "The Encryption request you specified is not valid. Supported value: AES256.",
            ArgumentName="x-amz-server-side-encryption",
            ArgumentValue=algorithm,
        )

    sse_customer_key = base64.b64decode(encryption_key)
    if len(sse_customer_key) != 32:
        raise InvalidArgument(
            "The secret key was invalid for the specified algorithm.",
            ArgumentName="x-amz-server-side-encryption",
            ArgumentValue="null",
        )

    sse_customer_key_md5 = base64.b64encode(hashlib.md5(sse_customer_key).digest()).decode("utf-8")
    if sse_customer_key_md5 != encryption_key_md5:
        raise InvalidArgument(
            "The calculated MD5 hash of the key did not match the hash that was provided.",
            # weirdly, the argument name is wrong, it should be `x-amz-server-side-encryption-customer-key-MD5`
            ArgumentName="x-amz-server-side-encryption",
            ArgumentValue="null",
        )


def validate_checksum_value(checksum_value: str, checksum_algorithm: ChecksumAlgorithm) -> bool:
    try:
        checksum = base64.b64decode(checksum_value)
    except Exception:
        return False

    match checksum_algorithm:
        case ChecksumAlgorithm.CRC32 | ChecksumAlgorithm.CRC32C:
            valid_length = 4
        case ChecksumAlgorithm.CRC64NVME:
            valid_length = 8
        case ChecksumAlgorithm.SHA1:
            valid_length = 20
        case ChecksumAlgorithm.SHA256:
            valid_length = 32
        case _:
            valid_length = 0

    return len(checksum) == valid_length

1	import base64	1✔
2	import datetime	1✔
3	import hashlib	1✔
4	from zoneinfo import ZoneInfo	1✔
5
6	from botocore.utils import InvalidArnException	1✔
7
8	from localstack.aws.api import CommonServiceException	1✔
9	from localstack.aws.api.s3 import (	1✔
10	AccessControlPolicy,
11	AnalyticsConfiguration,
12	AnalyticsId,
13	BucketCannedACL,
14	BucketLifecycleConfiguration,
15	BucketName,
16	ChecksumAlgorithm,
17	CORSConfiguration,
18	Grant,
19	Grantee,
20	Grants,
21	IntelligentTieringConfiguration,
22	IntelligentTieringId,
23	InvalidArgument,
24	InvalidBucketName,
25	InvalidEncryptionAlgorithmError,
26	InventoryConfiguration,
27	InventoryId,
28	KeyTooLongError,
29	ObjectCannedACL,
30	Permission,
31	ServerSideEncryption,
32	SSECustomerAlgorithm,
33	SSECustomerKey,
34	SSECustomerKeyMD5,
35	WebsiteConfiguration,
36	)
37	from localstack.aws.api.s3 import Type as GranteeType	1✔
38	from localstack.services.s3 import constants as s3_constants	1✔
39	from localstack.services.s3.exceptions import InvalidRequest, MalformedACLError, MalformedXML	1✔
40	from localstack.services.s3.utils import (	1✔
41	get_class_attrs_from_spec_class,
42	get_permission_header_name,
43	is_bucket_name_valid,
44	is_valid_canonical_id,
45	validate_dict_fields,
46	)
47	from localstack.utils.aws import arns	1✔
48	from localstack.utils.strings import to_bytes	1✔
49
50	# https://docs.aws.amazon.com/AmazonS3/latest/userguide/acl-overview.html#canned-acl
51	# bucket-owner-read + bucket-owner-full-control are allowed, but ignored for buckets
52	VALID_CANNED_ACLS = get_class_attrs_from_spec_class(	1✔
53	BucketCannedACL
54	) \| get_class_attrs_from_spec_class(ObjectCannedACL)
55
56
57	def validate_bucket_analytics_configuration(	1✔
58	id: AnalyticsId, analytics_configuration: AnalyticsConfiguration
59	) -> None:
60	if id != analytics_configuration.get("Id"):	1✔
61	raise MalformedXML(	1✔
62	"The XML you provided was not well-formed or did not validate against our published schema"
63	)
64
65
66	def validate_bucket_intelligent_tiering_configuration(	1✔
67	id: IntelligentTieringId, intelligent_tiering_configuration: IntelligentTieringConfiguration
68	) -> None:
69	if id != intelligent_tiering_configuration.get("Id"):	1✔
70	raise MalformedXML(	1✔
71	"The XML you provided was not well-formed or did not validate against our published schema"
72	)
73
74
75	def validate_bucket_name(bucket: BucketName) -> None:	1✔
76	"""
77	Validate s3 bucket name based on the documentation
78	ref. https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
79	"""
UNCOV 80	if not is_bucket_name_valid(bucket_name=bucket):	×
81	raise InvalidBucketName("The specified bucket is not valid.", BucketName=bucket)	×
82
83
84	def validate_canned_acl(canned_acl: str) -> None:	1✔
85	"""
86	Validate the canned ACL value, or raise an Exception
87	"""
88	if canned_acl and canned_acl not in VALID_CANNED_ACLS:	1✔
89	raise InvalidArgument(	1✔
90	None,
91	ArgumentName="x-amz-acl",
92	ArgumentValue=canned_acl,
93	)
94
95
96	def parse_grants_in_headers(permission: Permission, grantees: str) -> Grants:	1✔
97	splitted_grantees = [grantee.strip() for grantee in grantees.split(",")]	1✔
98	grants = []	1✔
99	for seralized_grantee in splitted_grantees:	1✔
100	grantee_type, grantee_id = seralized_grantee.split("=")	1✔
101	grantee_id = grantee_id.strip('"')	1✔
102	if grantee_type not in ("uri", "id", "emailAddress"):	1✔
103	raise InvalidArgument(	1✔
104	"Argument format not recognized",
105	ArgumentName=get_permission_header_name(permission),
106	ArgumentValue=seralized_grantee,
107	)
108	elif grantee_type == "uri":	1✔
109	if grantee_id not in s3_constants.VALID_ACL_PREDEFINED_GROUPS:	1✔
110	raise InvalidArgument(	1✔
111	"Invalid group uri",
112	ArgumentName="uri",
113	ArgumentValue=grantee_id,
114	)
115	grantee = Grantee(	1✔
116	Type=GranteeType.Group,
117	URI=grantee_id,
118	)
119
120	elif grantee_type == "id":	1✔
121	if not is_valid_canonical_id(grantee_id):	1✔
122	raise InvalidArgument(	1✔
123	"Invalid id",
124	ArgumentName="id",
125	ArgumentValue=grantee_id,
126	)
UNCOV 127	grantee = Grantee(	×
128	Type=GranteeType.CanonicalUser,
129	ID=grantee_id,
130	DisplayName="webfile", # TODO: only in certain regions
131	)
132
133	else:
134	# TODO: check validation here
UNCOV 135	grantee = Grantee(	×
136	Type=GranteeType.AmazonCustomerByEmail,
137	EmailAddress=grantee_id,
138	)
139	grants.append(Grant(Permission=permission, Grantee=grantee))	1✔
140
141	return grants	1✔
142
143
144	def validate_acl_acp(acp: AccessControlPolicy) -> None:	1✔
145	if acp is None or "Owner" not in acp or "Grants" not in acp:	1✔
146	raise MalformedACLError(	1✔
147	"The XML you provided was not well-formed or did not validate against our published schema"
148	)
149
150	if not is_valid_canonical_id(owner_id := acp["Owner"].get("ID", "")):	1✔
151	raise InvalidArgument(	1✔
152	"Invalid id",
153	ArgumentName="CanonicalUser/ID",
154	ArgumentValue=owner_id,
155	)
156
157	for grant in acp["Grants"]:	1✔
158	if grant.get("Permission") not in s3_constants.VALID_GRANTEE_PERMISSIONS:	1✔
159	raise MalformedACLError(	1✔
160	"The XML you provided was not well-formed or did not validate against our published schema"
161	)
162
163	grantee = grant.get("Grantee", {})	1✔
164	grant_type = grantee.get("Type")	1✔
165	if grant_type not in (	1✔
166	GranteeType.Group,
167	GranteeType.CanonicalUser,
168	GranteeType.AmazonCustomerByEmail,
169	):
170	raise MalformedACLError(	1✔
171	"The XML you provided was not well-formed or did not validate against our published schema"
172	)
173	elif (	1✔
174	grant_type == GranteeType.Group
175	and (grant_uri := grantee.get("URI", ""))
176	not in s3_constants.VALID_ACL_PREDEFINED_GROUPS
177	):
178	raise InvalidArgument(	1✔
179	"Invalid group uri",
180	ArgumentName="Group/URI",
181	ArgumentValue=grant_uri,
182	)
183
184	elif grant_type == GranteeType.AmazonCustomerByEmail:	1✔
185	# TODO: add validation here
UNCOV 186	continue	×
187
188	elif grant_type == GranteeType.CanonicalUser and not is_valid_canonical_id(	1✔
189	grantee_id := grantee.get("ID", "")
190	):
191	raise InvalidArgument(	1✔
192	"Invalid id",
193	ArgumentName="CanonicalUser/ID",
194	ArgumentValue=grantee_id,
195	)
196
197
198	def validate_lifecycle_configuration(lifecycle_conf: BucketLifecycleConfiguration) -> None:	1✔
199	"""
200	Validate the Lifecycle configuration following AWS docs
201	See https://docs.aws.amazon.com/AmazonS3/latest/userguide/intro-lifecycle-rules.html
202	https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketLifecycleConfiguration.html
203	:param lifecycle_conf: the bucket lifecycle configuration given by the client
204	:raises MalformedXML: when the file doesn't follow the basic structure/required fields
205	:raises InvalidArgument: if the `Date` passed for the Expiration is not at Midnight GMT
206	:raises InvalidRequest: if there are duplicate tags keys in `Tags` field
207	:return: None
208	"""
209	# we only add the `Expiration` header, we don't delete objects yet
210	# We don't really expire or transition objects
211	# TODO: transition not supported not validated, as we don't use it yet
212	if not lifecycle_conf:	1✔
UNCOV 213	return	×
214
215	for rule in lifecycle_conf.get("Rules", []):	1✔
216	if any(req_key not in rule for req_key in ("ID", "Filter", "Status")):	1✔
217	raise MalformedXML()	1✔
218	if (non_current_exp := rule.get("NoncurrentVersionExpiration")) is not None:	1✔
219	if all(	1✔
220	req_key not in non_current_exp
221	for req_key in ("NewerNoncurrentVersions", "NoncurrentDays")
222	):
223	raise MalformedXML()	1✔
224
225	if rule_filter := rule.get("Filter"):	1✔
226	if len(rule_filter) > 1:	1✔
227	raise MalformedXML()	1✔
228
229	if (expiration := rule.get("Expiration", {})) and "ExpiredObjectDeleteMarker" in expiration:	1✔
230	if len(expiration) > 1:	1✔
231	raise MalformedXML()	1✔
232
233	if exp_date := (expiration.get("Date")):	1✔
234	if exp_date.timetz() != datetime.time(	1✔
235	hour=0, minute=0, second=0, microsecond=0, tzinfo=ZoneInfo("GMT")
236	):
237	raise InvalidArgument(	1✔
238	"'Date' must be at midnight GMT",
239	ArgumentName="Date",
240	ArgumentValue=exp_date.astimezone(), # use the locale timezone, that's what AWS does (returns PST?)
241	)
242
243	if tags := (rule_filter.get("And", {}).get("Tags")):	1✔
244	tag_keys = set()	1✔
245	for tag in tags:	1✔
246	if (tag_key := tag.get("Key")) in tag_keys:	1✔
247	raise InvalidRequest("Duplicate Tag Keys are not allowed.")	1✔
248	tag_keys.add(tag_key)	1✔
249
250
251	def validate_website_configuration(website_config: WebsiteConfiguration) -> None:	1✔
252	"""
253	Validate the website configuration following AWS docs
254	See https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketWebsite.html
255	:param website_config:
256	:raises
257	:return: None
258	"""
259	if redirect_all_req := website_config.get("RedirectAllRequestsTo", {}):	1✔
260	if len(website_config) > 1:	1✔
261	raise InvalidArgument(	1✔
262	"RedirectAllRequestsTo cannot be provided in conjunction with other Routing Rules.",
263	ArgumentName="RedirectAllRequestsTo",
264	ArgumentValue="not null",
265	)
266
267	if "HostName" not in redirect_all_req:	1✔
UNCOV 268	raise MalformedXML()	×
269
270	if (protocol := redirect_all_req.get("Protocol")) and protocol not in ("http", "https"):	1✔
UNCOV 271	raise InvalidRequest(	×
272	"Invalid protocol, protocol can be http or https. If not defined the protocol will be selected automatically."
273	)
274
275	return	1✔
276
277	# required
278	# https://docs.aws.amazon.com/AmazonS3/latest/API/API_IndexDocument.html
279	if not (index_configuration := website_config.get("IndexDocument")):	1✔
280	raise InvalidArgument(	1✔
281	"A value for IndexDocument Suffix must be provided if RedirectAllRequestsTo is empty",
282	ArgumentName="IndexDocument",
283	ArgumentValue="null",
284	)
285
286	if not (index_suffix := index_configuration.get("Suffix")) or "/" in index_suffix:	1✔
287	raise InvalidArgument(	1✔
288	"The IndexDocument Suffix is not well formed",
289	ArgumentName="IndexDocument",
290	ArgumentValue=index_suffix or None,
291	)
292
293	if "ErrorDocument" in website_config and not website_config.get("ErrorDocument", {}).get("Key"):	1✔
UNCOV 294	raise MalformedXML()	×
295
296	if "RoutingRules" in website_config:	1✔
297	routing_rules = website_config.get("RoutingRules", [])	1✔
298	if len(routing_rules) == 0:	1✔
299	raise MalformedXML()	1✔
300	if len(routing_rules) > 50:	1✔
UNCOV 301	raise ValueError("Too many routing rules") # TODO: correct exception	×
302	for routing_rule in routing_rules:	1✔
303	redirect = routing_rule.get("Redirect", {})	1✔
304	# todo: this does not raise an error? check what GetWebsiteConfig returns? empty field?
305	# if not (redirect := routing_rule.get("Redirect")):
306	# raise "Something"
307
308	if "ReplaceKeyPrefixWith" in redirect and "ReplaceKeyWith" in redirect:	1✔
309	raise InvalidRequest(	1✔
310	"You can only define ReplaceKeyPrefix or ReplaceKey but not both."
311	)
312
313	if "Condition" in routing_rule and not routing_rule.get("Condition", {}):	1✔
314	raise InvalidRequest(	1✔
315	"Condition cannot be empty. To redirect all requests without a condition, the condition element shouldn't be present."
316	)
317
318	if (protocol := redirect.get("Protocol")) and protocol not in ("http", "https"):	1✔
319	raise InvalidRequest(	1✔
320	"Invalid protocol, protocol can be http or https. If not defined the protocol will be selected automatically."
321	)
322
323
324	def validate_inventory_configuration(	1✔
325	config_id: InventoryId, inventory_configuration: InventoryConfiguration
326	):
327	"""
328	Validate the Inventory Configuration following AWS docs
329	Validation order is XML then `Id` then S3DestinationBucket
330	https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketInventoryConfiguration.html
331	https://docs.aws.amazon.com/AmazonS3/latest/userguide/storage-inventory.html
332	:param config_id: the passed Id parameter passed to the provider method
333	:param inventory_configuration: InventoryConfiguration
334	:raises MalformedXML: when the file doesn't follow the basic structure/required fields
335	:raises IdMismatch: if the `Id` parameter is different from the `Id` field from the configuration
336	:raises InvalidS3DestinationBucket: if S3 bucket is not provided as an ARN
337	:return: None
338	"""
339	required_root_fields = {"Destination", "Id", "IncludedObjectVersions", "IsEnabled", "Schedule"}	1✔
340	optional_root_fields = {"Filter", "OptionalFields"}	1✔
341
342	if not validate_dict_fields(	1✔
343	inventory_configuration, required_root_fields, optional_root_fields
344	):
UNCOV 345	raise MalformedXML()	×
346
347	required_s3_bucket_dest_fields = {"Bucket", "Format"}	1✔
348	optional_s3_bucket_dest_fields = {"AccountId", "Encryption", "Prefix"}	1✔
349
350	if not (	1✔
351	s3_bucket_destination := inventory_configuration["Destination"].get("S3BucketDestination")
352	) or not validate_dict_fields(
353	s3_bucket_destination, required_s3_bucket_dest_fields, optional_s3_bucket_dest_fields
354	):
UNCOV 355	raise MalformedXML()	×
356
357	if inventory_configuration["Destination"]["S3BucketDestination"]["Format"] not in (	1✔
358	"CSV",
359	"ORC",
360	"Parquet",
361	):
362	raise MalformedXML()	1✔
363
364	if not (frequency := inventory_configuration["Schedule"].get("Frequency")) or frequency not in (	1✔
365	"Daily",
366	"Weekly",
367	):
368	raise MalformedXML()	1✔
369
370	if inventory_configuration["IncludedObjectVersions"] not in ("All", "Current"):	1✔
371	raise MalformedXML()	1✔
372
373	possible_optional_fields = {	1✔
374	"Size",
375	"LastModifiedDate",
376	"StorageClass",
377	"ETag",
378	"IsMultipartUploaded",
379	"ReplicationStatus",
380	"EncryptionStatus",
381	"ObjectLockRetainUntilDate",
382	"ObjectLockMode",
383	"ObjectLockLegalHoldStatus",
384	"IntelligentTieringAccessTier",
385	"BucketKeyStatus",
386	"ChecksumAlgorithm",
387	}
388	if (opt_fields := inventory_configuration.get("OptionalFields")) and set(	1✔
389	opt_fields
390	) - possible_optional_fields:
391	raise MalformedXML()	1✔
392
393	if inventory_configuration.get("Id") != config_id:	1✔
394	raise CommonServiceException(	1✔
395	code="IdMismatch", message="Document ID does not match the specified configuration ID."
396	)
397
398	bucket_arn = inventory_configuration["Destination"]["S3BucketDestination"]["Bucket"]	1✔
399	try:	1✔
400	arns.parse_arn(bucket_arn)	1✔
401	except InvalidArnException:	1✔
402	raise CommonServiceException(	1✔
403	code="InvalidS3DestinationBucket", message="Invalid bucket ARN."
404	)
405
406
407	def validate_cors_configuration(cors_configuration: CORSConfiguration):	1✔
408	rules = cors_configuration["CORSRules"]	1✔
409
410	if not rules or len(rules) > 100:	1✔
411	raise MalformedXML()	1✔
412
413	required_rule_fields = {"AllowedMethods", "AllowedOrigins"}	1✔
414	optional_rule_fields = {"AllowedHeaders", "ExposeHeaders", "MaxAgeSeconds", "ID"}	1✔
415
416	for rule in rules:	1✔
417	if not validate_dict_fields(rule, required_rule_fields, optional_rule_fields):	1✔
UNCOV 418	raise MalformedXML()	×
419
420	for method in rule["AllowedMethods"]:	1✔
421	if method not in ("GET", "PUT", "HEAD", "POST", "DELETE"):	1✔
422	raise InvalidRequest(	1✔
423	f"Found unsupported HTTP method in CORS config. Unsupported method is {method}"
424	)
425
426
427	def validate_object_key(object_key: str) -> None:	1✔
428	"""
429	ref. https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-keys.html
430
431	"""
432	if (len_key := len(to_bytes(object_key, encoding="UTF-8"))) > 1024:	1✔
UNCOV 433	raise KeyTooLongError(	×
434	"Your key is too long",
435	MaxSizeAllowed="1024",
436	Size=str(len_key),
437	)
438
439
440	def validate_sse_c(	1✔
441	algorithm: SSECustomerAlgorithm,
442	encryption_key: SSECustomerKey,
443	encryption_key_md5: SSECustomerKeyMD5,
444	server_side_encryption: ServerSideEncryption = None,
445	):
446	"""
447	This method validates the SSE Customer parameters for different requests.
448	:param algorithm: the SSECustomerAlgorithm parameter of the incoming Request, can only be AES256
449	:param encryption_key: the SSECustomerKey of the incoming Request, represent the base64 encoded encryption key
450	:param encryption_key_md5: the SSECustomerKeyMD5 of the request, represents the base64 encoded MD5 hash of the
451	encryption key
452	:param server_side_encryption: when the incoming request is a "write" request (PutObject, CopyObject,
453	CreateMultipartUpload), the user can specify the encryption. Customer encryption and AWS SSE can't both be set.
454	:raises: InvalidArgument if the request is invalid
455	:raises: InvalidEncryptionAlgorithmError if the given algorithm is different from AES256
456	"""
457	if not encryption_key and not algorithm:	1✔
458	return	1✔
459	elif server_side_encryption:	1✔
460	raise InvalidArgument(	1✔
461	"Server Side Encryption with Customer provided key is incompatible with the encryption method specified",
462	ArgumentName="x-amz-server-side-encryption",
463	ArgumentValue=server_side_encryption,
464	)
465
466	if encryption_key and not algorithm:	1✔
467	raise InvalidArgument(	1✔
468	"Requests specifying Server Side Encryption with Customer provided keys must provide a valid encryption algorithm.",
469	ArgumentName="x-amz-server-side-encryption",
470	ArgumentValue="null",
471	)
472	elif not encryption_key and algorithm:	1✔
473	raise InvalidArgument(	1✔
474	"Requests specifying Server Side Encryption with Customer provided keys must provide an appropriate secret key.",
475	ArgumentName="x-amz-server-side-encryption",
476	ArgumentValue="null",
477	)
478
479	if algorithm != "AES256":	1✔
480	raise InvalidEncryptionAlgorithmError(	1✔
481	"The Encryption request you specified is not valid. Supported value: AES256.",
482	ArgumentName="x-amz-server-side-encryption",
483	ArgumentValue=algorithm,
484	)
485
486	sse_customer_key = base64.b64decode(encryption_key)	1✔
487	if len(sse_customer_key) != 32:	1✔
488	raise InvalidArgument(	1✔
489	"The secret key was invalid for the specified algorithm.",
490	ArgumentName="x-amz-server-side-encryption",
491	ArgumentValue="null",
492	)
493
494	sse_customer_key_md5 = base64.b64encode(hashlib.md5(sse_customer_key).digest()).decode("utf-8")	1✔
495	if sse_customer_key_md5 != encryption_key_md5:	1✔
496	raise InvalidArgument(	1✔
497	"The calculated MD5 hash of the key did not match the hash that was provided.",
498	# weirdly, the argument name is wrong, it should be `x-amz-server-side-encryption-customer-key-MD5`
499	ArgumentName="x-amz-server-side-encryption",
500	ArgumentValue="null",
501	)
502
503
504	def validate_checksum_value(checksum_value: str, checksum_algorithm: ChecksumAlgorithm) -> bool:	1✔
505	try:	1✔
506	checksum = base64.b64decode(checksum_value)	1✔
507	except Exception:	1✔
508	return False	1✔
509
510	match checksum_algorithm:	1✔
511	case ChecksumAlgorithm.CRC32 \| ChecksumAlgorithm.CRC32C:	1✔
512	valid_length = 4	1✔
513	case ChecksumAlgorithm.CRC64NVME:	1✔
514	valid_length = 8	1✔
515	case ChecksumAlgorithm.SHA1:	1✔
516	valid_length = 20	1✔
517	case ChecksumAlgorithm.SHA256:	1✔
518	valid_length = 32	1✔
UNCOV 519	case _:	×
UNCOV 520	valid_length = 0	×
521
522	return len(checksum) == valid_length	1✔

localstack / localstack / 17481980735

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