localstack · bentsku · Sep 4, 2025 · Sep 2, 2025 · Sep 2, 2025 · Sep 2, 2025
@@ -21,6 +21,7 @@
     Event,
     EventBridgeConfiguration,
     EventList,
+    InvalidArgument,
     LambdaFunctionArn,
     LambdaFunctionConfiguration,
     NotificationConfiguration,
@@ -34,8 +35,8 @@
     TopicConfiguration,
 )
 from localstack.aws.connect import connect_to
+from localstack.services.s3.exceptions import MalformedXML
 from localstack.services.s3.models import S3Bucket, S3DeleteMarker, S3Object
-from localstack.services.s3.utils import _create_invalid_argument_exc
 from localstack.utils.aws import arns
 from localstack.utils.aws.arns import ARN_PARTITION_REGEX, get_partition, parse_arn, s3_bucket_arn
 from localstack.utils.aws.client_types import ServicePrincipal
@@ -272,26 +273,29 @@ def _validate_notification(self, verification_ctx: BucketVerificationContext):
         arn, argument_name = self._get_arn_value_and_name(configuration)
 
         if not re.match(f"{ARN_PARTITION_REGEX}:{self.service_name}:", arn):
-            raise _create_invalid_argument_exc(
-                "The ARN could not be parsed", name=argument_name, value=arn
+            raise InvalidArgument(
+                "The ARN could not be parsed",
+                ArgumentName=argument_name,
+                ArgumentValue=arn,
             )
+
         if not verification_ctx.skip_destination_validation:
             self._verify_target(arn, verification_ctx)
 
         if filter_rules := configuration.get("Filter", {}).get("Key", {}).get("FilterRules"):
             for rule in filter_rules:
-                rule["Name"] = rule["Name"].capitalize()
-                if rule["Name"] not in ["Suffix", "Prefix"]:
-                    raise _create_invalid_argument_exc(
+                if "Name" not in rule or "Value" not in rule:
+                    raise MalformedXML()
+
+                if rule["Name"].lower() not in ["suffix", "prefix"]:
+                    raise InvalidArgument(
                         "filter rule name must be either prefix or suffix",
-                        rule["Name"],
-                        rule["Value"],
-                    )
-                if not rule["Value"]:
-                    raise _create_invalid_argument_exc(
-                        "filter value cannot be empty", rule["Name"], rule["Value"]
+                        ArgumentName="FilterRule.Name",
+                        ArgumentValue=rule["Name"],
                     )
 
+                rule["Name"] = rule["Name"].capitalize()
+
     @staticmethod
     def _get_test_payload(verification_ctx: BucketVerificationContext):
         return {
@@ -382,7 +386,7 @@ class SqsNotifier(BaseNotifier):
 
     @staticmethod
     def _get_arn_value_and_name(queue_configuration: QueueConfiguration) -> tuple[QueueArn, str]:
-        return queue_configuration.get("QueueArn", ""), "QueueArn"
+        return queue_configuration.get("QueueArn", ""), "Queue"
 
     def _verify_target(self, target_arn: str, verification_ctx: BucketVerificationContext) -> None:
         if not is_api_enabled("sqs"):
@@ -410,11 +414,12 @@ def _verify_target(self, target_arn: str, verification_ctx: BucketVerificationCo
                 code,
                 exc_info=LOG.isEnabledFor(logging.DEBUG),
             )
-            raise _create_invalid_argument_exc(
+            raise InvalidArgument(
                 "Unable to validate the following destination configurations",
-                name=target_arn,
-                value="The destination queue does not exist",
+                ArgumentName=target_arn,
+                ArgumentValue="The destination queue does not exist",
             )
+
         # send test event with the request metadata for permissions
         # https://docs.aws.amazon.com/AmazonS3/latest/userguide/notification-how-to-event-types-and-destinations.html#supported-notification-event-types
         sqs_client = connect_to(region_name=arn_data["region"]).sqs.request_metadata(
@@ -430,10 +435,10 @@ def _verify_target(self, target_arn: str, verification_ctx: BucketVerificationCo
                 verification_ctx.bucket_name,
                 target_arn,
             )
-            raise _create_invalid_argument_exc(
+            raise InvalidArgument(
                 "Unable to validate the following destination configurations",
-                name=target_arn,
-                value="Permissions on the destination queue do not allow S3 to publish notifications from this bucket",
+                ArgumentName=target_arn,
+                ArgumentValue="Permissions on the destination queue do not allow S3 to publish notifications from this bucket",
             ) from e
 
     def notify(self, ctx: S3EventNotificationContext, config: QueueConfiguration):
@@ -490,10 +495,10 @@ def _verify_target(self, target_arn: str, verification_ctx: BucketVerificationCo
         try:
             sns_client.get_topic_attributes(TopicArn=target_arn)
         except ClientError:
-            raise _create_invalid_argument_exc(
+            raise InvalidArgument(
                 "Unable to validate the following destination configurations",
-                name=target_arn,
-                value="The destination topic does not exist",
+                ArgumentName=target_arn,
+                ArgumentValue="The destination topic does not exist",
             )
 
         sns_client = connect_to(region_name=arn_data["region"]).sns.request_metadata(
@@ -513,10 +518,10 @@ def _verify_target(self, target_arn: str, verification_ctx: BucketVerificationCo
                 verification_ctx.bucket_name,
                 target_arn,
             )
-            raise _create_invalid_argument_exc(
+            raise InvalidArgument(
                 "Unable to validate the following destination configurations",
-                name=target_arn,
-                value="Permissions on the destination topic do not allow S3 to publish notifications from this bucket",
+                ArgumentName=target_arn,
+                ArgumentValue="Permissions on the destination topic do not allow S3 to publish notifications from this bucket",
             ) from e
 
     def notify(self, ctx: S3EventNotificationContext, config: TopicConfiguration):
@@ -575,10 +580,10 @@ def _verify_target(self, target_arn: str, verification_ctx: BucketVerificationCo
         try:
             lambda_client.get_function(FunctionName=target_arn)
         except ClientError:
-            raise _create_invalid_argument_exc(
+            raise InvalidArgument(
                 "Unable to validate the following destination configurations",
-                name=target_arn,
-                value="The destination Lambda does not exist",
+                ArgumentName=target_arn,
+                ArgumentValue="The destination Lambda does not exist",
             )
         lambda_client = connect_to(region_name=arn_data["region"]).lambda_.request_metadata(
             source_arn=s3_bucket_arn(verification_ctx.bucket_name, region=verification_ctx.region),
@@ -587,10 +592,10 @@ def _verify_target(self, target_arn: str, verification_ctx: BucketVerificationCo
         try:
             lambda_client.invoke(FunctionName=target_arn, InvocationType=InvocationType.DryRun)
         except ClientError as e:
-            raise _create_invalid_argument_exc(
+            raise InvalidArgument(
                 "Unable to validate the following destination configurations",
-                name=f"{target_arn}, null",
-                value=f"Not authorized to invoke function [{target_arn}]",
+                ArgumentName=f"{target_arn}, null",
+                ArgumentValue=f"Not authorized to invoke function [{target_arn}]",
             ) from e
 
     def notify(self, ctx: S3EventNotificationContext, config: LambdaFunctionConfiguration):

@@ -45,7 +45,6 @@
 )
 from localstack.services.s3.utils import (
     S3_VIRTUAL_HOST_FORWARDED_HEADER,
-    _create_invalid_argument_exc,
     capitalize_header_name_from_snake_case,
     extract_bucket_name_and_key_from_headers_and_path,
     forwarded_from_virtual_host_addressed_request,
@@ -772,13 +771,12 @@ def validate_post_policy(
     :return: None
     """
     if not request_form.get("key"):
-        ex: InvalidArgument = _create_invalid_argument_exc(
-            message="Bucket POST must contain a field named 'key'.  If it is specified, please check the order of the fields.",
-            name="key",
-            value="",
-            host_id=FAKE_HOST_ID,
+        raise InvalidArgument(
+            "Bucket POST must contain a field named 'key'.  If it is specified, please check the order of the fields.",
+            ArgumentName="key",
+            ArgumentValue="",
+            HostId=FAKE_HOST_ID,
         )
-        raise ex
 
     form_dict = {k.lower(): v for k, v in request_form.items()}
 
@@ -932,13 +930,12 @@ def _is_match_with_signature_fields(
                 if argument_name == "Awsaccesskeyid":
                     argument_name = "AWSAccessKeyId"
 
-                ex: InvalidArgument = _create_invalid_argument_exc(
-                    message=f"Bucket POST must contain a field named '{argument_name}'.  If it is specified, please check the order of the fields.",
-                    name=argument_name,
-                    value="",
-                    host_id=FAKE_HOST_ID,
+                raise InvalidArgument(
+                    f"Bucket POST must contain a field named '{argument_name}'.  If it is specified, please check the order of the fields.",
+                    ArgumentName=argument_name,
+                    ArgumentValue="",
+                    HostId=FAKE_HOST_ID,
                 )
-                raise ex
 
         return True
     return False
@@ -574,17 +574,6 @@ def get_bucket_and_key_from_presign_url(presign_url: str) -> tuple[str, str]:
     return bucket, key
 
 
-def _create_invalid_argument_exc(
-    message: str | None, name: str, value: str, host_id: str = None
-) -> InvalidArgument:
-    ex = InvalidArgument(message)
-    ex.ArgumentName = name
-    ex.ArgumentValue = value
-    if host_id:
-        ex.HostId = host_id
-    return ex
-
-
 def capitalize_header_name_from_snake_case(header_name: str) -> str:
     return "-".join([part.capitalize() for part in header_name.split("-")])
 

@@ -38,7 +38,6 @@
 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,
@@ -87,8 +86,11 @@ 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
+        raise InvalidArgument(
+            None,
+            ArgumentName="x-amz-acl",
+            ArgumentValue=canned_acl,
+        )
 
 
 def parse_grants_in_headers(permission: Permission, grantees: str) -> Grants:
@@ -98,25 +100,30 @@ def parse_grants_in_headers(permission: Permission, grantees: str) -> Grants:
         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(
+            raise InvalidArgument(
                 "Argument format not recognized",
-                get_permission_header_name(permission),
-                seralized_grantee,
+                ArgumentName=get_permission_header_name(permission),
+                ArgumentValue=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
+                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):
-                ex = _create_invalid_argument_exc("Invalid id", "id", grantee_id)
-                raise ex
+                raise InvalidArgument(
+                    "Invalid id",
+                    ArgumentName="id",
+                    ArgumentValue=grantee_id,
+                )
             grantee = Grantee(
                 Type=GranteeType.CanonicalUser,
                 ID=grantee_id,
@@ -141,8 +148,11 @@ def validate_acl_acp(acp: AccessControlPolicy) -> None:
         )
 
     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
+        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:
@@ -165,8 +175,11 @@ def validate_acl_acp(acp: AccessControlPolicy) -> None:
             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
+            raise InvalidArgument(
+                "Invalid group uri",
+                ArgumentName="Group/URI",
+                ArgumentValue=grant_uri,
+            )
 
         elif grant_type == GranteeType.AmazonCustomerByEmail:
             # TODO: add validation here
@@ -175,8 +188,11 @@ def validate_acl_acp(acp: AccessControlPolicy) -> None:
         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
+            raise InvalidArgument(
+                "Invalid id",
+                ArgumentName="CanonicalUser/ID",
+                ArgumentValue=grantee_id,
+            )
 
 
 def validate_lifecycle_configuration(lifecycle_conf: BucketLifecycleConfiguration) -> None:
@@ -242,12 +258,12 @@ def validate_website_configuration(website_config: WebsiteConfiguration) -> 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 InvalidArgument(
+                "RedirectAllRequestsTo cannot be provided in conjunction with other Routing Rules.",
+                ArgumentName="RedirectAllRequestsTo",
+                ArgumentValue="not null",
             )
-            raise ex
+
         if "HostName" not in redirect_all_req:
             raise MalformedXML()
 
@@ -261,20 +277,18 @@ def validate_website_configuration(website_config: WebsiteConfiguration) -> None
     # 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 InvalidArgument(
+            "A value for IndexDocument Suffix must be provided if RedirectAllRequestsTo is empty",
+            ArgumentName="IndexDocument",
+            ArgumentValue="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 InvalidArgument(
+            "The IndexDocument Suffix is not well formed",
+            ArgumentName="IndexDocument",
+            ArgumentValue=index_suffix or None,
         )
-        raise ex
 
     if "ErrorDocument" in website_config and not website_config.get("ErrorDocument", {}).get("Key"):
         raise MalformedXML()