Step Function State Machine: Implement check for StartAt

Author: r-heimann

src/cfnlint/rules/resources/stepfunctions/StateMachineDefinition.py

@@ -103,6 +103,61 @@ def _clean_schema(self, validator: Validator, instance: Any):
             if ql == "JSONata":
                 yield self._convert_schema_to_jsonata(), ql_validator
 
+    def _validate_start_at(
+        self,
+        definition: Any,
+        k: str,
+        add_path_to_message: bool,
+        path: deque | None = None,
+    ) -> ValidationResult:
+        """
+        Per the Amazon States Language specification, 'StartAt must' reference
+        a valid state name that exists in the States object.
+
+        Reference: https://states-language.net/spec.html#toplevelfields
+        """
+
+        start_at = definition.get("StartAt")
+        states = definition.get("States")
+
+        # Check if StartAt is missing
+        if not isinstance(start_at, str) and not isinstance(states, dict):
+            return  # Early return to avoid further checks
+
+        # Check if StartAt state exists in States object
+        if start_at not in states:
+            if path is None:  # Top level StartAt
+                error_path = deque([k, "StartAt"])
+                display_path = "/StartAt"
+            else:  # Nested StartAt like Parallel or Map
+                error_path = deque([k] + list(path) + ["StartAt"])
+                display_path = f"/{'/'.join(str(item) for item in path)}/StartAt"
+
+            message = f"Missing 'Next' target '{start_at}' at {display_path}"
+
+            yield ValidationError(message, path=error_path, rule=self)
+
+        # Validate nested StartAt in Parallel and Map states
+        for state_name, state in states.items():
+            state_type = state.get("Type")
+
+            if state_type == "Parallel":
+                branches = state.get("Branches", [])
+                if not isinstance(branches, list):
+                    continue
+                for idx, branch in enumerate(branches):
+                    branch_path = deque(["States", state_name, "Branches", idx])
+                    yield from self._validate_start_at(
+                        branch, k, add_path_to_message, branch_path
+                    )
+
+            if state_type == "Map":
+                processor = state.get("ItemProcessor")
+                processor_path = deque(["States", state_name, "ItemProcessor"])
+                yield from self._validate_start_at(
+                    processor, k, add_path_to_message, processor_path
+                )
+
     def _validate_step(
         self,
         validator: Validator,
@@ -134,6 +189,9 @@ def _validate_step(
 
             yield self._clean_error(err)
 
+        # Validate StartAt exists
+        yield from self._validate_start_at(value, k, add_path_to_message)
+
     def validate(
         self, validator: Validator, keywords: Any, instance: Any, schema: dict[str, Any]
     ) -> ValidationResult:

test/unit/rules/resources/stepfunctions/test_state_machine_definition.py

@@ -1354,6 +1354,103 @@ def rule():
                 ),
             ],
         ),
+        (
+            "Missing StartAt target",
+            {
+                "Definition": {
+                    "StartAt": "FAIL",
+                    "States": {
+                        "Pass": {
+                            "Type": "Pass",
+                            "Next": "Success",
+                        },
+                        "Success": {
+                            "Type": "Succeed",
+                        },
+                    },
+                }
+            },
+            [
+                ValidationError(
+                    "Missing 'Next' target 'FAIL' at /StartAt",
+                    rule=StateMachineDefinition(),
+                    path=deque(["Definition", "StartAt"]),
+                ),
+            ],
+        ),
+        (
+            "Parallel state with missing StartAt target",
+            {
+                "Definition": {
+                    "StartAt": "ParallelState",
+                    "States": {
+                        "ParallelState": {
+                            "Type": "Parallel",
+                            "Branches": [
+                                {
+                                    "StartAt": "FAIL",
+                                    "States": {
+                                        "BranchState": {
+                                            "Type": "Pass",
+                                            "End": True,
+                                        },
+                                    },
+                                }
+                            ],
+                            "End": True,
+                        },
+                    },
+                }
+            },
+            [
+                ValidationError(
+                    "Missing 'Next' target 'FAIL' at /States/ParallelState/Branches/0/StartAt",
+                    rule=StateMachineDefinition(),
+                    path=deque(
+                        [
+                            "Definition",
+                            "States",
+                            "ParallelState",
+                            "Branches",
+                            0,
+                            "StartAt",
+                        ]
+                    ),
+                ),
+            ],
+        ),
+        (
+            "Map state with missing StartAt target in ItemProcessor",
+            {
+                "Definition": {
+                    "StartAt": "MapState",
+                    "States": {
+                        "MapState": {
+                            "Type": "Map",
+                            "ItemProcessor": {
+                                "StartAt": "FAIL",
+                                "States": {
+                                    "ProcessItem": {
+                                        "Type": "Pass",
+                                        "End": True,
+                                    },
+                                },
+                            },
+                            "End": True,
+                        },
+                    },
+                }
+            },
+            [
+                ValidationError(
+                    "Missing 'Next' target 'FAIL' at /States/MapState/ItemProcessor/StartAt",
+                    rule=StateMachineDefinition(),
+                    path=deque(
+                        ["Definition", "States", "MapState", "ItemProcessor", "StartAt"]
+                    ),
+                ),
+            ],
+        ),
     ],
 )
 def test_validate(