anchorectl: add format check (#12375)

* anchorectl: add format check

* anchorectl: add format check

Author: valentijnscholten

dojo/tools/anchorectl_policies/parser.py

@@ -28,53 +28,54 @@ def get_findings(self, filename, test):
 
         find_date = datetime.now()
         items = []
-        try:
-            for image in data:
-                if image["detail"] is not None:
-                    for result in image["detail"]:
-                        try:
-                            gate = result["gate"]
-                            description = result["description"]
-                            policy_id = result["policyId"]
-                            status = result["status"]
-                            image_name = result["tag"]
-                            trigger_id = result["triggerId"]
-                            repo, tag = image_name.split(":", 2)
-                            severity, active = get_severity(status, description)
-                            vulnerability_id = extract_vulnerability_id(trigger_id)
-                            title = (
-                                policy_id
-                                + " - gate|"
-                                + gate
-                                + " - trigger|"
-                                + trigger_id
-                            )
-                            find = Finding(
-                                title=title,
-                                test=test,
-                                description=description,
-                                severity=severity,
-                                active=active,
-                                references=f"Policy ID: {policy_id}\nTrigger ID: {trigger_id}",
-                                file_path=search_filepath(description),
-                                component_name=repo,
-                                component_version=tag,
-                                date=find_date,
-                                static_finding=True,
-                                dynamic_finding=False,
-                            )
-                            if vulnerability_id:
-                                find.unsaved_vulnerability_ids = [vulnerability_id]
-                            items.append(find)
-                        except (KeyError, IndexError) as err:
-                            msg = f"Invalid format: {err} key not found"
-                            raise ValueError(msg)
-        except AttributeError as err:
-            # import empty policies without error (e.g. policies or images
-            # objects are not a dictionary)
-            logger.warning(
-                "Exception at %s", "parsing anchore policy", exc_info=err,
-            )
+
+        if not isinstance(data, list):
+            msg = "This doesn't look like a valid Anchore CTRL Policies report: Expected a list with image data at the root of the JSON data"
+            raise TypeError(msg)
+
+        for image in data:
+            if not isinstance(image, dict) or image.get("detail") is None or not isinstance(image.get("detail"), list):
+                msg = "This doesn't look like a valid Anchore CTRL Policies report, missing 'detail' list object key for image"
+                raise ValueError(msg)
+
+            for result in image["detail"]:
+                try:
+                    gate = result["gate"]
+                    description = result["description"]
+                    policy_id = result["policyId"]
+                    status = result["status"]
+                    image_name = result["tag"]
+                    trigger_id = result["triggerId"]
+                    repo, tag = image_name.split(":", 2)
+                    severity, active = get_severity(status, description)
+                    vulnerability_id = extract_vulnerability_id(trigger_id)
+                    title = (
+                        policy_id
+                        + " - gate|"
+                        + gate
+                        + " - trigger|"
+                        + trigger_id
+                    )
+                    find = Finding(
+                        title=title,
+                        test=test,
+                        description=description,
+                        severity=severity,
+                        active=active,
+                        references=f"Policy ID: {policy_id}\nTrigger ID: {trigger_id}",
+                        file_path=search_filepath(description),
+                        component_name=repo,
+                        component_version=tag,
+                        date=find_date,
+                        static_finding=True,
+                        dynamic_finding=False,
+                    )
+                    if vulnerability_id:
+                        find.unsaved_vulnerability_ids = [vulnerability_id]
+                    items.append(find)
+                except (KeyError, IndexError) as err:
+                    msg = f"Invalid format: {err} key not found"
+                    raise ValueError(msg)
         return items
 
 

unittests/scans/anchorectl_policies/many_violations.json

@@ -40,5 +40,4 @@
       "status": "fail",
       "tag": "test/testimage:testtag"
     }
-  ]
-  
+  ]