Add new rules for SQS queue

src/cfnlint/data/schemas/extensions/aws_sqs_queue/properties.json

@@ -0,0 +1,57 @@
+{
+ "allOf": [
+  {
+   "if": {
+    "properties": {
+     "FifoQueue": {
+      "enum": [
+       true,
+       "true"
+      ]
+     }
+    },
+    "required": [
+     "FifoQueue"
+    ]
+   },
+   "then": {
+    "if": {
+     "properties": {
+      "QueueName": {
+       "type": "string"
+      }
+     },
+     "required": [
+      "QueueName"
+     ]
+    },
+    "then": {
+     "properties": {
+      "QueueName": {
+       "pattern": "^.*\\.fifo$"
+      }
+     }
+    }
+   }
+  },
+  {
+   "if": {
+    "properties": {
+     "FifoQueue": {
+      "enum": [
+       false,
+       "false"
+      ]
+     }
+    }
+   },
+   "then": {
+    "properties": {
+     "ContentBasedDeduplication": false,
+     "DeduplicationScope": false,
+     "FifoThroughputLimit": false
+    }
+   }
+  }
+ ]
+}

src/cfnlint/data/schemas/patches/providers/all/aws_sqs_queue/redrivepolicy.json

@@ -0,0 +1,19 @@
+[
+ {
+  "op": "add",
+  "path": "/properties/RedrivePolicy/additionalProperties",
+  "value": false
+ },
+ {
+  "op": "add",
+  "path": "/properties/RedrivePolicy/properties",
+  "value": {
+   "deadLetterTargetArn": {
+    "type": "string"
+   },
+   "maxReceiveCount": {
+    "type": "integer"
+   }
+  }
+ }
+]

src/cfnlint/data/schemas/providers/us_east_1/aws-sqs-queue.json

@@ -82,10 +82,16 @@
    ]
   },
   "RedrivePolicy": {
-   "type": [
-    "object",
-    "string"
-   ]
+   "additionalProperties": false,
+   "properties": {
+    "deadLetterTargetArn": {
+     "type": "string"
+    },
+    "maxReceiveCount": {
+     "type": "integer"
+    }
+   },
+   "type": "object"
   },
   "SqsManagedSseEnabled": {
    "type": "boolean"

src/cfnlint/rules/resources/sqs/QueueDLQ.py

@@ -0,0 +1,101 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from collections import deque
+from typing import Any, Iterator
+
+import cfnlint.data.schemas.extensions.aws_sqs_queue
+from cfnlint.helpers import bool_compare, is_function
+from cfnlint.jsonschema import ValidationError, ValidationResult, Validator
+from cfnlint.rules.helpers import get_value_from_path
+from cfnlint.rules.jsonschema.CfnLintJsonSchema import CfnLintJsonSchema, SchemaDetails
+
+
+class QueueDLQ(CfnLintJsonSchema):
+    id = "E3502"
+    shortdesc = "Validate SQS queue properties are valid"
+    description = (
+        "Depending on if the queue is FIFO or not the properties and "
+        "allowed values change. This rule validates properties and "
+        "values based on the queue type"
+    )
+    source_url = "https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-sqs-queue.html"
+    tags = ["resources", "sqs"]
+
+    def __init__(self) -> None:
+        super().__init__(
+            keywords=[
+                "Resources/AWS::SQS::Queue/Properties",
+            ],
+            schema_details=SchemaDetails(
+                module=cfnlint.data.schemas.extensions.aws_sqs_queue,
+                filename="properties.json",
+            ),
+            all_matches=True,
+        )
+
+    def _is_fifo_queue(
+        self, validator: Validator, instance: Any
+    ) -> Iterator[tuple[str, Validator]]:
+        standard = "standard"
+        fifo = "FIFO"
+
+        if "FifoQueue" not in instance:
+            yield standard, validator
+            return
+
+        for queue_type, queue_type_validator in get_value_from_path(
+            validator=validator, instance=instance, path=deque(["FifoQueue"])
+        ):
+            yield (
+                fifo if bool_compare(queue_type, True) else standard
+            ), queue_type_validator
+
+    def validate(
+        self, validator: Validator, _: Any, instance: Any, schema: dict[str, Any]
+    ) -> ValidationResult:
+
+        for queue_type, queue_type_validator in self._is_fifo_queue(
+            validator=validator,
+            instance=instance,
+        ):
+            queue_type_validator = queue_type_validator.evolve(
+                context=queue_type_validator.context.evolve(
+                    path=validator.context.path.evolve()
+                )
+            )
+
+            for target, target_validator in get_value_from_path(
+                queue_type_validator,
+                instance,
+                path=deque(["RedrivePolicy", "deadLetterTargetArn"]),
+            ):
+                k, v = is_function(target)
+                if k == "Fn::GetAtt":
+                    if target_validator.is_type(v, "string"):
+                        v = v.split(".")
+
+                    if len(v) < 1:
+                        return
+
+                    dest_queue = validator.cfn.template.get("Resources", {}).get(
+                        v[0], {}
+                    )
+                    if dest_queue.get("Type") != "AWS::SQS::Queue":
+                        return
+
+                    for dest_queue_type, _ in self._is_fifo_queue(
+                        target_validator,
+                        instance=dest_queue.get("Properties", {}),
+                    ):
+                        if queue_type != dest_queue_type:
+                            yield ValidationError(
+                                (
+                                    f"Source queue type {queue_type!r} does not "
+                                    "match destination queue type {dest_queue_type!r}"
+                                ),
+                                rule=self,
+                                path=target_validator.context.path.path,
+                            )

src/cfnlint/rules/resources/sqs/QueueProperties.py

@@ -0,0 +1,46 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from typing import Any
+
+import cfnlint.data.schemas.extensions.aws_sqs_queue
+from cfnlint.jsonschema import ValidationResult, Validator
+from cfnlint.rules.jsonschema.CfnLintJsonSchema import CfnLintJsonSchema, SchemaDetails
+
+
+class QueueProperties(CfnLintJsonSchema):
+    id = "E3501"
+    shortdesc = "Validate SQS queue properties are valid"
+    description = (
+        "Depending on if the queue is FIFO or not the "
+        "properties and allowed values change. "
+        "This rule validates properties and values based on the queue type"
+    )
+    source_url = "https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-sqs-queue.html"
+    tags = ["resources", "sqs"]
+
+    def __init__(self) -> None:
+        super().__init__(
+            keywords=[
+                "Resources/AWS::SQS::Queue/Properties",
+            ],
+            schema_details=SchemaDetails(
+                module=cfnlint.data.schemas.extensions.aws_sqs_queue,
+                filename="properties.json",
+            ),
+            all_matches=True,
+        )
+
+    def validate(
+        self, validator: Validator, keywords: Any, instance: Any, schema: Any
+    ) -> ValidationResult:
+        for err in super().validate(validator, keywords, instance, schema):
+            if err.schema is False:
+                err.message = (
+                    f"Additional properties are not allowed ({err.path[-1]!r} "
+                    "was unexpected)"
+                )
+
+            yield err