311e4cb7-85c6-4815-9a05-f76a2f16b041

Committed 27 Jan 2025 05:40PM UTC coverage: 86.9% (-0.004%) from 86.904%

Build # 311e4cb7-85c6-4815-9a05-f76a2f16b041

Build Type

push

circleci

Committed by

web-flow

Commit Message

Fix: validate schedule_expression for EventBridge Scheduler (#12191)

Coverage Stats

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

127 existing lines in 14 files now uncovered.

61173 of 70395 relevant lines covered (86.9%)

0.87 hits per line

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

92.0

/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 (
    _create_invalid_argument_exc,
    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:
        ex = _create_invalid_argument_exc(None, "x-amz-acl", canned_acl)
        raise ex


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"):
            ex = _create_invalid_argument_exc(
                "Argument format not recognized",
                get_permission_header_name(permission),
                seralized_grantee,
            )
            raise ex
        elif grantee_type == "uri":
            if grantee_id not in s3_constants.VALID_ACL_PREDEFINED_GROUPS:
                ex = _create_invalid_argument_exc("Invalid group uri", "uri", grantee_id)
                raise ex
            grantee = Grantee(
                Type=GranteeType.Group,
                URI=grantee_id,
            )

        elif grantee_type == "id":
            if not is_valid_canonical_id(grantee_id):
                ex = _create_invalid_argument_exc("Invalid id", "id", grantee_id)
                raise ex
            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", "")):
        ex = _create_invalid_argument_exc("Invalid id", "CanonicalUser/ID", owner_id)
        raise ex

    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
        ):
            ex = _create_invalid_argument_exc("Invalid group uri", "Group/URI", grant_uri)
            raise ex

        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", ""))
        ):
            ex = _create_invalid_argument_exc("Invalid id", "CanonicalUser/ID", grantee_id)
            raise ex


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:
            ex = _create_invalid_argument_exc(
                message="RedirectAllRequestsTo cannot be provided in conjunction with other Routing Rules.",
                name="RedirectAllRequestsTo",
                value="not null",
            )
            raise ex
        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")):
        ex = _create_invalid_argument_exc(
            message="A value for IndexDocument Suffix must be provided if RedirectAllRequestsTo is empty",
            name="IndexDocument",
            value="null",
        )
        raise ex

    if not (index_suffix := index_configuration.get("Suffix")) or "/" in index_suffix:
        ex = _create_invalid_argument_exc(
            message="The IndexDocument Suffix is not well formed",
            name="IndexDocument",
            value=index_suffix or None,
        )
        raise ex

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

localstack / localstack / 311e4cb7-85c6-4815-9a05-f76a2f16b041

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