22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

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

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

20.73

/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,
    EncodingType,
    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",
        )
    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",
        )

    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",
        )

    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",
        )


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


def validate_encoding_type(encoding_type: EncodingType):
    if encoding_type is not None and not encoding_type == EncodingType.url:
        raise InvalidArgument(
            "Invalid Encoding Method specified in Request",
            ArgumentName="encoding-type",
            ArgumentValue=encoding_type,
        )

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

localstack / localstack / 22709357475

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