cowork-bot: harden parser error-handling for malformed HCL/CloudFormation/Pulumi

.github/workflows/ci.yml

@@ -17,7 +17,7 @@ jobs:
         python-version: ["3.10", "3.11", "3.12", "3.13"]
 
     steps:
-      - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
+      - uses: actions/checkout@v4
         with:
           persist-credentials: false
 

.github/workflows/publish.yml

@@ -21,7 +21,7 @@ jobs:
       id-token: write
 
     steps:
-      - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
+      - uses: actions/checkout@v4
         with:
           persist-credentials: false
 

.gitignore

@@ -87,3 +87,6 @@ local.db
 
 # Added by release-prep
 node_modules
+
+# npm lock artifact (Python-only project)
+package-lock.json

src/deploydiff/cloudformation_parser.py

@@ -52,13 +52,24 @@ def parse_cloudformation_changeset(changeset_json: str | dict[str, Any]) -> Depl
     else:
         data = changeset_json
 
+    if not isinstance(data, dict):
+        raise ValueError("Change set input must be a JSON object")
+
     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

@@ -50,16 +50,23 @@ def parse_pulumi_preview(preview_json: str | dict[str, Any]) -> DeployPlan:
     else:
         data = preview_json
 
+    if not isinstance(data, dict):
+        raise ValueError("Preview input must be a JSON object")
+
     changes: list[ResourceChange] = []
 
     # 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"))
 
@@ -97,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

@@ -44,15 +44,32 @@ def parse_terraform_plan(plan_json: str | dict[str, Any]) -> DeployPlan:
     else:
         data = plan_json
 
+    if not isinstance(data, dict):
+        raise ValueError("Plan input must be a JSON object")
+
     format_version = data.get("format_version", "")
     changes: list[ResourceChange] = []
 
     # 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

@@ -59,3 +59,49 @@ def test_cloudformation_valid_dict_still_works(self):
         data = {"Changes": []}
         plan = parse_cloudformation_changeset(data)
         assert len(plan.changes) == 0
+
+    @pytest.mark.parametrize(
+        ("parser", "payload"),
+        [
+            (parse_terraform_plan, []),
+            (parse_cloudformation_changeset, []),
+            (parse_pulumi_preview, []),
+        ],
+    )
+    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}}})