cowork-bot: harden parser error-handling for malformed inputs

Author: Coding-Dev-Tools

src/deploydiff/cloudformation_parser.py

@@ -57,11 +57,19 @@ def parse_cloudformation_changeset(changeset_json: str | dict[str, Any]) -> Depl
 
     changes: list[ResourceChange] = []
     changes_list = data.get("Changes", data.get("changes", []))
+    if not isinstance(changes_list, list):
+        raise ValueError("CloudFormation Changes must be a JSON array")
 
-    for change_entry in changes_list:
+    for index, change_entry in enumerate(changes_list):
+        if not isinstance(change_entry, dict):
+            raise ValueError(f"CloudFormation Changes[{index}] must be a JSON object")
         resource_change_data = change_entry.get(
             "ResourceChange", change_entry.get("resource_change", {})
         )
+        if not isinstance(resource_change_data, dict):
+            raise ValueError(
+                f"CloudFormation Changes[{index}].ResourceChange must be a JSON object"
+            )
         action_str = change_entry.get(
             "Action", resource_change_data.get("Action", "Modify")
         )

src/deploydiff/pulumi_parser.py

@@ -57,12 +57,16 @@ def parse_pulumi_preview(preview_json: str | dict[str, Any]) -> DeployPlan:
 
     # Pulumi preview JSON has a "steps" array
     steps = data.get("steps", [])
+    if not isinstance(steps, list):
+        raise ValueError("Pulumi steps must be a JSON array")
 
     # Also support the resource-oriented format
     resources = data.get("resourceChanges", data.get("resources", {}))
 
     # Process steps-based format
-    for step in steps:
+    for index, step in enumerate(steps):
+        if not isinstance(step, dict):
+            raise ValueError(f"Pulumi steps[{index}] must be a JSON object")
         urn = step.get("urn", step.get("old", {}).get("urn", "unknown"))
         step_type = step.get("step", step.get("op", "same"))
 
@@ -100,9 +104,19 @@ def parse_pulumi_preview(preview_json: str | dict[str, Any]) -> DeployPlan:
         changes.append(resource_change)
 
     # Process resource-changes-based format (count-based)
-    if not steps and isinstance(resources, dict):
+    if not steps:
+        if not isinstance(resources, dict):
+            raise ValueError("Pulumi resourceChanges must be a JSON object")
         for resource_type, counts in resources.items():
+            if not isinstance(counts, dict):
+                raise ValueError(
+                    f"Pulumi resourceChanges[{resource_type!r}] must be a JSON object"
+                )
             for action_str, count in counts.items():
+                if not isinstance(count, int) or isinstance(count, bool) or count < 0:
+                    raise ValueError(
+                        f"Pulumi resourceChanges[{resource_type!r}][{action_str!r}] must be a non-negative integer"
+                    )
                 action = PULUMI_STEP_MAP.get(action_str, ChangeAction.UPDATE)
                 for i in range(count):
                     resource_change = ResourceChange(

src/deploydiff/terraform_parser.py

@@ -52,10 +52,24 @@ def parse_terraform_plan(plan_json: str | dict[str, Any]) -> DeployPlan:
 
     # Parse planned changes
     resource_changes = data.get("resource_changes", [])
+    if not isinstance(resource_changes, list):
+        raise ValueError("Terraform resource_changes must be a JSON array")
 
-    for rc in resource_changes:
+    for index, rc in enumerate(resource_changes):
+        if not isinstance(rc, dict):
+            raise ValueError(f"Terraform resource_changes[{index}] must be a JSON object")
         change = rc.get("change", {})
+        if not isinstance(change, dict):
+            raise ValueError(
+                f"Terraform resource_changes[{index}].change must be a JSON object"
+            )
         action_strs = change.get("actions", [])
+        if not isinstance(action_strs, list) or not all(
+            isinstance(action, str) for action in action_strs
+        ):
+            raise ValueError(
+                f"Terraform resource_changes[{index}].change.actions must be a JSON array of strings"
+            )
 
         # Use the primary action
         primary_action = _resolve_primary_action(action_strs)

tests/test_parse_errors.py

@@ -72,3 +72,36 @@ def test_json_array_is_rejected_with_clear_error(self, parser, payload):
         """A decoded JSON value must be an object before parser-specific access."""
         with pytest.raises(ValueError, match="JSON object"):
             parser(payload)
+
+    @pytest.mark.parametrize(
+        ("parser", "payload", "message"),
+        [
+            (
+                parse_terraform_plan,
+                {"resource_changes": {}},
+                "resource_changes must be a JSON array",
+            ),
+            (
+                parse_cloudformation_changeset,
+                {"Changes": {}},
+                "Changes must be a JSON array",
+            ),
+            (
+                parse_pulumi_preview,
+                {"steps": {}},
+                "steps must be a JSON array",
+            ),
+        ],
+    )
+    def test_malformed_collections_raise_clear_error(self, parser, payload, message):
+        """Malformed collection fields must not be silently ignored."""
+        with pytest.raises(ValueError, match=message):
+            parser(payload)
+
+    def test_terraform_malformed_resource_entry_is_rejected(self):
+        with pytest.raises(ValueError, match=r"resource_changes\[0\].*JSON object"):
+            parse_terraform_plan({"resource_changes": ["not-an-object"]})
+
+    def test_pulumi_negative_resource_count_is_rejected(self):
+        with pytest.raises(ValueError, match="non-negative integer"):
+            parse_pulumi_preview({"resourceChanges": {"aws:s3/bucket:Bucket": {"create": -1}}})