Repository: aws-cloudformation/cfn-lint

Issue:
Step Functions JSONata isn't fully supported - "Arguments" and "QueryLanguage" properties are not allowed
### CloudFormation Lint Version

1.22.4

### What operating system are you using?

MacOS

### Describe the bug

AWS [not long ago](https://aws.amazon.com/blogs/compute/simplifying-developer-experience-with-variables-and-jsonata-in-aws-step-functions/) introduced JSONata support in Step Functions.

CFN-Lint keeps throwing errors when using JSONata within the SF state machine's definition properly:
```
E3601     Additional properties are not allowed ('Arguments' was unexpected)
E3601     Additional properties are not allowed ('QueryLanguage' was unexpected)
```

The template below does work in CloudFormation itself of course.

Relates also to Issue #3858


### Expected behavior

CFN-Lint will not detect JSONata syntax in SF as errors.

### Reproduction template

```yaml
Resources:
  LogGroup:
    Type: AWS::Logs::LogGroup
    UpdateReplacePolicy: Delete
    DeletionPolicy: Delete
    Properties:
      RetentionInDays: 14

  StateMachinee5ae0411:
    Type: AWS::StepFunctions::StateMachine
    Properties:
      Definition:
        StartAt: CreateQueue
        States:
          CreateQueue:
            Next: DeleteQueue
            Resource: arn:aws:states:::aws-sdk:sqs:createQueue
            Type: Task
            Assign:
              inputPayload: '{% $states.context.Execution.Input %}'
              QueueUrl: '{% $states.result.QueueUrl %}'
            QueryLanguage: JSONata
            Arguments:
              Attributes:
                MessageRetentionPeriod: '345600'
                ReceiveMessageWaitTimeSeconds: '20'
                VisibilityTimeout: '120'
              QueueName: '{% ''prefixxxxxx-'' & $states.input.bucketName %}'
              Tags:
                Bucket: '{% $states.input.bucketName %}'
          DeleteQueue:
            Type: Task
            Resource: arn:aws:states:::aws-sdk:sqs:deleteQueue
            Next: Success
            QueryLanguage: JSONata
            Arguments:
              QueueUrl: '{% $QueueUrl %}'
          Success:
            Type: Succeed
      StateMachineName: StateMachinee5ae0411
      StateMachineType: STANDARD
      RoleArn: "arn:aws:blababla"
      EncryptionConfiguration:
        Type: AWS_OWNED_KEY
      LoggingConfiguration:
        Level: ERROR
        IncludeExecutionData: true
        Destinations:
          - CloudWatchLogsLogGroup:
              LogGroupArn: !GetAtt LogGroup.Arn
```

Discussion:
@kddejong: Fix is merged into main and I'll try to release today.@tomholmes4: @kddejong Could we also add support for the Condition attribute inside of Choice states please? 

Solution patch:
diff --git a/src/cfnlint/data/schemas/other/step_functions/statemachine.json b/src/cfnlint/data/schemas/other/step_functions/statemachine.json
index 9acb8ac0e..b33f1803d 100644
--- a/src/cfnlint/data/schemas/other/step_functions/statemachine.json
+++ b/src/cfnlint/data/schemas/other/step_functions/statemachine.json
@@ -2,6 +2,7 @@
  "additionalProperties": false,
  "definitions": {
   "choice": {
+   "$ref": "#/definitions/common",
    "additionalProperties": false,
    "definitions": {
     "Assign": {
@@ -197,6 +198,9 @@
     }
    },
    "properties": {
+    "Arguments": {
+     "type": "object"
+    },
     "Choices": {
      "items": {
       "$ref": "#/definitions/choice/definitions/Operator"
@@ -224,12 +228,21 @@
      "pattern": "^.{1,128}$",
      "type": "string"
     },
+    "Output": {
+     "type": [
+      "string",
+      "object"
+     ]
+    },
     "OutputPath": {
      "type": [
       "string",
       "null"
      ]
     },
+    "QueryLanguage": {
+     "$ref": "#/definitions/queryLanguage"
+    },
     "Type": {
      "enum": [
       "Choice"
@@ -243,9 +256,67 @@
    ],
    "type": "object"
   },
+  "common": {
+   "allOf": [
+    {
+     "if": {
+      "properties": {
+       "QueryLanguage": {
+        "const": "JSONata"
+       }
+      },
+      "required": [
+       "QueryLanguage"
+      ]
+     },
+     "then": {
+      "properties": {
+       "InputPath": false,
+       "OutputPath": false,
+       "Parameters": false,
+       "ResultPath": false,
+       "ResultSelector": false
+      }
+     }
+    },
+    {
+     "if": {
+      "oneOf": [
+       {
+        "properties": {
+         "QueryLanguage": {
+          "const": "JSONPath"
+         }
+        },
+        "required": [
+         "QueryLanguage"
+        ]
+       },
+       {
+        "not": {
+         "required": [
+          "QueryLanguage"
+         ]
+        }
+       }
+      ]
+     },
+     "then": {
+      "properties": {
+       "Arguments": false,
+       "Output": false
+      }
+     }
+    }
+   ]
+  },
   "fail": {
+   "$ref": "#/definitions/common",
    "additionalProperties": false,
    "properties": {
+    "Arguments": {
+     "type": "object"
+    },
     "Cause": {
      "type": "string"
     },
@@ -267,12 +338,21 @@
       "null"
      ]
     },
+    "Output": {
+     "type": [
+      "string",
+      "object"
+     ]
+    },
     "OutputPath": {
      "type": [
       "string",
       "null"
      ]
     },
+    "QueryLanguage": {
+     "$ref": "#/definitions/queryLanguage"
+    },
     "Type": {
      "enum": [
       "Fail"
@@ -286,8 +366,12 @@
    "type": "object"
   },
   "map": {
+   "$ref": "#/definitions/common",
    "additionalProperties": false,
    "properties": {
+    "Arguments": {
+     "type": "object"
+    },
     "Assign": {
      "type": "object"
     },
@@ -394,6 +478,12 @@
      "pattern": "^.{1,128}$",
      "type": "string"
     },
+    "Output": {
+     "type": [
+      "string",
+      "object"
+     ]
+    },
     "OutputPath": {
      "type": [
       "string",
@@ -403,6 +493,9 @@
     "Parameters": {
      "type": "object"
     },
+    "QueryLanguage": {
+     "$ref": "#/definitions/queryLanguage"
+    },
     "ResultPath": {
      "type": [
       "string",
@@ -474,8 +567,12 @@
    "type": "object"
   },
   "parallel": {
+   "$ref": "#/definitions/common",
    "additionalProperties": false,
    "properties": {
+    "Arguments": {
+     "type": "object"
+    },
     "Assign": {
      "type": "object"
     },
@@ -525,12 +622,21 @@
      "pattern": "^.{1,128}$",
      "type": "string"
     },
+    "Output": {
+     "type": [
+      "string",
+      "object"
+     ]
+    },
     "OutputPath": {
      "type": [
       "string",
       "null"
      ]
     },
+    "QueryLanguage": {
+     "$ref": "#/definitions/queryLanguage"
+    },
     "ResultPath": {
      "type": [
       "string",
@@ -587,8 +693,12 @@
    "type": "object"
   },
   "pass": {
+   "$ref": "#/definitions/common",
    "additionalProperties": false,
    "properties": {
+    "Arguments": {
+     "type": "object"
+    },
     "Assign": {
      "type": "object"
     },
@@ -610,6 +720,12 @@
      "pattern": "^.{1,128}$",
      "type": "string"
     },
+    "Output": {
+     "type": [
+      "string",
+      "object"
+     ]
+    },
     "OutputPath": {
      "type": [
       "string",
@@ -619,6 +735,9 @@
     "Parameters": {
      "type": "object"
     },
+    "QueryLanguage": {
+     "$ref": "#/definitions/queryLanguage"
+    },
     "Result": {},
     "ResultPath": {
      "type": "string"
@@ -639,6 +758,14 @@
    ],
    "type": "object"
   },
+  "queryLanguage": {
+   "default": "JSONPath",
+   "enum": [
+    "JSONPath",
+    "JSONata"
+   ],
+   "type": "string"
+  },
   "state": {
    "allOf": [
     {
@@ -804,8 +931,12 @@
    "type": "object"
   },
   "task": {
+   "$ref": "#/definitions/common",
    "additionalProperties": false,
    "properties": {
+    "Arguments": {
+     "type": "object"
+    },
     "Assign": {
      "type": "object"
     },
@@ -862,6 +993,12 @@
      "pattern": "^.{1,128}$",
      "type": "string"
     },
+    "Output": {
+     "type": [
+      "string",
+      "object"
+     ]
+    },
     "OutputPath": {
      "type": [
       "string",
@@ -871,6 +1008,9 @@
     "Parameters": {
      "type": "object"
     },
+    "QueryLanguage": {
+     "$ref": "#/definitions/queryLanguage"
+    },
     "Resource": {
      "pattern": "^arn:aws:([a-z]|-)+:([a-z]|[0-9]|-)*:[0-9]*:([a-z]|-)+:[a-zA-Z0-9-_.]+(:(\\$LATEST|[a-zA-Z0-9-_\\.]+))?$",
      "type": [
@@ -940,8 +1080,12 @@
    "type": "object"
   },
   "wait": {
+   "$ref": "#/definitions/common",
    "additionalProperties": false,
    "properties": {
+    "Arguments": {
+     "type": "object"
+    },
     "Assign": {
      "type": "object"
     },
@@ -963,12 +1107,21 @@
      "pattern": "^.{1,128}$",
      "type": "string"
     },
+    "Output": {
+     "type": [
+      "string",
+      "object"
+     ]
+    },
     "OutputPath": {
      "type": [
       "string",
       "null"
      ]
     },
+    "QueryLanguage": {
+     "$ref": "#/definitions/queryLanguage"
+    },
     "Seconds": {
      "minimum": 0,
      "type": "number"
@@ -1009,6 +1162,9 @@
   "Comment": {
    "type": "string"
   },
+  "QueryLanguage": {
+   "$ref": "#/definitions/queryLanguage"
+  },
   "StartAt": {
    "type": "string"
   },
diff --git a/src/cfnlint/rules/resources/stepfunctions/StateMachineDefinition.py b/src/cfnlint/rules/resources/stepfunctions/StateMachineDefinition.py
index a06e34a59..34616fe6d 100644
--- a/src/cfnlint/rules/resources/stepfunctions/StateMachineDefinition.py
+++ b/src/cfnlint/rules/resources/stepfunctions/StateMachineDefinition.py
@@ -107,9 +107,15 @@ class StateMachineDefinition(CfnLintJsonSchema):
                     err = self._fix_message(err)
 
                 err.path.appendleft(k)
-                if not err.validator.startswith("fn_") and err.validator not in [
-                    "cfnLint"
-                ]:
+                if err.schema in [True, False]:
+                    err.message = (
+                        f"Additional properties are not allowed ({err.path[-1]!r} "
+                        "was unexpected)"
+                    )
+                if not err.validator or (
+                    not err.validator.startswith("fn_")
+                    and err.validator not in ["cfnLint"]
+                ):
                     err.rule = self
 
                 yield self._clean_error(err)


Test patch:
diff --git a/test/unit/rules/resources/stepfunctions/test_state_machine_definition.py b/test/unit/rules/resources/stepfunctions/test_state_machine_definition.py
index 954fd4a85..49f5dc412 100644
--- a/test/unit/rules/resources/stepfunctions/test_state_machine_definition.py
+++ b/test/unit/rules/resources/stepfunctions/test_state_machine_definition.py
@@ -694,6 +694,171 @@ def rule():
             {"Definition": {"Fn::Join": ["\n", []]}},
             [],
         ),
+        (
+            "Invalid query languages",
+            {
+                "DefinitionSubstitutions": {"jobqueue": {"Ref": "MyJob"}},
+                "Definition": {
+                    "Comment": (
+                        "An example of the Amazon States "
+                        "Language for notification on an "
+                        "AWS Batch job completion"
+                    ),
+                    "StartAt": "Submit Batch Job",
+                    "TimeoutSeconds": 3600,
+                    "States": {
+                        "Submit Batch Job 1": {
+                            "Type": "Task",
+                            "Resource": "${jobqueue}",
+                            "Parameters": {  # fails because JSONata doesn't support Parameters
+                                "JobName": "BatchJobNotification",
+                                "JobQueue": "arn:aws:batch:us-east-1:123456789012:job-queue/BatchJobQueue-7049d367474b4dd",
+                                "JobDefinition": "arn:aws:batch:us-east-1:123456789012:job-definition/BatchJobDefinition-74d55ec34c4643c:1",
+                            },
+                            "Next": "Notify Success",
+                            "Catch": [
+                                {
+                                    "ErrorEquals": ["States.ALL"],
+                                    "Next": "Notify Failure",
+                                }
+                            ],
+                            "QueryLanguage": "JSONata",
+                        },
+                        "Submit Batch Job 2": {
+                            "Type": "Task",
+                            "Resource": "${jobqueue}",
+                            "Arguments": {  # fails because JSONPath doesn't support Arguments
+                                "JobName": "BatchJobNotification",
+                                "JobQueue": "arn:aws:batch:us-east-1:123456789012:job-queue/BatchJobQueue-7049d367474b4dd",
+                                "JobDefinition": "arn:aws:batch:us-east-1:123456789012:job-definition/BatchJobDefinition-74d55ec34c4643c:1",
+                            },
+                            "Next": "Notify Success",
+                            "Catch": [
+                                {
+                                    "ErrorEquals": ["States.ALL"],
+                                    "Next": "Notify Failure",
+                                }
+                            ],
+                            "QueryLanguage": "JSONPath",
+                        },
+                        "Submit Batch Job 3": {
+                            "Type": "Task",
+                            "Resource": "${jobqueue}",
+                            "Arguments": {  # fails because JSONPath is the default for Query Language
+                                "JobName": "BatchJobNotification",
+                                "JobQueue": "arn:aws:batch:us-east-1:123456789012:job-queue/BatchJobQueue-7049d367474b4dd",
+                                "JobDefinition": "arn:aws:batch:us-east-1:123456789012:job-definition/BatchJobDefinition-74d55ec34c4643c:1",
+                            },
+                            "Next": "Notify Success",
+                            "Catch": [
+                                {
+                                    "ErrorEquals": ["States.ALL"],
+                                    "Next": "Notify Failure",
+                                }
+                            ],
+                        },
+                        "Notify Success": {
+                            "Type": "Task",
+                            "Resource": "arn:aws:states:::sns:publish",
+                            "Parameters": {
+                                "Message": "Batch job submitted through Step Functions succeeded",
+                                "TopicArn": "arn:aws:sns:us-east-1:123456789012:batchjobnotificatiointemplate-SNSTopic-1J757CVBQ2KHM",
+                            },
+                            "End": True,
+                        },
+                        "Notify Failure": {
+                            "Type": "Task",
+                            "Resource": "arn:aws:states:::sns:publish",
+                            "Parameters": {
+                                "Message": "Batch job submitted through Step Functions failed",
+                                "TopicArn": "arn:aws:sns:us-east-1:123456789012:batchjobnotificatiointemplate-SNSTopic-1J757CVBQ2KHM",
+                            },
+                            "End": True,
+                        },
+                    },
+                },
+            },
+            [
+                ValidationError(
+                    (
+                        "Additional properties are not allowed ('Parameters' was unexpected)"
+                    ),
+                    rule=StateMachineDefinition(),
+                    validator=None,
+                    schema_path=deque(
+                        [
+                            "properties",
+                            "States",
+                            "patternProperties",
+                            "^.{1,128}$",
+                            "allOf",
+                            5,
+                            "then",
+                            "allOf",
+                            0,
+                            "then",
+                            "properties",
+                            "Parameters",
+                        ]
+                    ),
+                    path=deque(
+                        ["Definition", "States", "Submit Batch Job 1", "Parameters"]
+                    ),
+                ),
+                ValidationError(
+                    (
+                        "Additional properties are not allowed ('Arguments' was unexpected)"
+                    ),
+                    rule=StateMachineDefinition(),
+                    validator=None,
+                    schema_path=deque(
+                        [
+                            "properties",
+                            "States",
+                            "patternProperties",
+                            "^.{1,128}$",
+                            "allOf",
+                            5,
+                            "then",
+                            "allOf",
+                            1,
+                            "then",
+                            "properties",
+                            "Arguments",
+                        ]
+                    ),
+                    path=deque(
+                        ["Definition", "States", "Submit Batch Job 2", "Arguments"]
+                    ),
+                ),
+                ValidationError(
+                    (
+                        "Additional properties are not allowed ('Arguments' was unexpected)"
+                    ),
+                    rule=StateMachineDefinition(),
+                    validator=None,
+                    schema_path=deque(
+                        [
+                            "properties",
+                            "States",
+                            "patternProperties",
+                            "^.{1,128}$",
+                            "allOf",
+                            5,
+                            "then",
+                            "allOf",
+                            1,
+                            "then",
+                            "properties",
+                            "Arguments",
+                        ]
+                    ),
+                    path=deque(
+                        ["Definition", "States", "Submit Batch Job 3", "Arguments"]
+                    ),
+                ),
+            ],
+        ),
     ],
 )
 def test_validate(