Add new "evaluations" format support to Anchorectl parser (#12425)

* Fix AnchoreCTL Policies parser to support new format with evaluations array

This commit updates the AnchoreCTL Policies parser to support both the legacy and new format reports generated by the AnchoreCTL tool.

Changes:
- Added detection for the new format which has an object with evaluations array instead of a root-level list
- Implemented conversion logic to transform the new format into a compatible structure for parsing
- Improved error handling with more descriptive messages
- Made field extraction more robust with proper fallbacks between formats

The parser now successfully processes both:
- Legacy format (list at root level)
- New format from anchorectl policy evaluate -o json (object with evaluations array)

* Added tests for the new format to verify correct parsing

* Fixed linter errors

* Update AnchoreCTL Policies Report documentation for clarity and format support

* Removed unnecessary text from anchorectl_policies

Author: cosmel-dojo

docs/content/en/connecting_your_tools/parsers/file/anchorectl_policies.md

@@ -2,7 +2,16 @@
 title: "AnchoreCTL Policies Report"
 toc_hide: true
 ---
-AnchoreCTLs JSON policies report format
+AnchoreCTLs JSON policies report format. Both legacy list-based format and new evaluation-based format are supported.
+
+## Usage
+
+To generate a policy report that can be imported into DefectDojo:
+
+```bash
+# Evaluate policies and output to JSON format
+anchorectl policy evaluate -o json > policy_report.json
+```
 
 ### Sample Scan Data
 Sample AnchoreCTL Policies Report scans can be found [here](https://github.com/DefectDojo/django-DefectDojo/tree/master/unittests/scans/anchorectl_policies).

dojo/tools/anchorectl_policies/parser.py

@@ -17,7 +17,7 @@ def get_label_for_scan_types(self, scan_type):
         return "AnchoreCTL Policies Report"
 
     def get_description_for_scan_types(self, scan_type):
-        return "AnchoreCTLs JSON policies report format."
+        return "AnchoreCTLs JSON policies report format. Both legacy list-based format and new evaluation-based format (from anchorectl policy evaluate -o json) are supported."
 
     def get_findings(self, filename, test):
         content = filename.read()
@@ -29,33 +29,81 @@ def get_findings(self, filename, test):
         find_date = datetime.now()
         items = []
 
+        # Handle new AnchoreCTL format (object with evaluations array)
+        if isinstance(data, dict) and "evaluations" in data:
+            logger.info("Detected new AnchoreCTL policies format")
+            processed_data = []
+            # Process each evaluation in the evaluations array
+            for evaluation in data.get("evaluations", []):
+                # Only process evaluations with findings
+                if evaluation.get("numberOfFindings", 0) > 0 and evaluation.get("details"):
+                    processed_item = {
+                        "detail": [],
+                        "digest": data.get("imageDigest", ""),
+                        "finalAction": evaluation.get("finalAction", ""),
+                        "finalActionReason": evaluation.get("finalActionReason", ""),
+                        "lastEvaluation": evaluation.get("evaluationTime", ""),
+                        "policyId": data.get("policyId", ""),
+                        "status": evaluation.get("status", ""),
+                        "tag": data.get("evaluatedTag", ""),
+                    }
+
+                    # Process details if they exist
+                    for detail in evaluation.get("details", []):
+                        processed_item["detail"].append(detail)
+
+                    processed_data.append(processed_item)
+
+            data = processed_data
+
         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"
+            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 or an object with 'evaluations' array"
             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"
+            # Skip empty images
+            if len(data) == 0:
+                continue
+
+            # Check for valid structure
+            if not isinstance(image, dict):
+                msg = "This doesn't look like a valid Anchore CTRL Policies report, expected dict object for image"
+                raise TypeError(msg)
+
+            # Handle legacy format with detail field
+            if image.get("detail") is not None and isinstance(image.get("detail"), list):
+                details = image.get("detail")
+            # Handle newer format that might have details under a different structure
+            elif image.get("details") is not None and isinstance(image.get("details"), list):
+                details = image.get("details")
+            else:
+                msg = "This doesn't look like a valid Anchore CTRL Policies report, missing 'detail' or 'details' list object key for image"
                 raise ValueError(msg)
 
-            for result in image["detail"]:
+            # Process each finding detail
+            for result in details:
                 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)
+                    # Extract fields with fallbacks for different formats
+                    gate = result.get("gate", "unknown")
+                    description = result.get("description", "No description provided")
+                    policy_id = result.get("policyId", image.get("policyId", "unknown"))
+                    status = result.get("status", "unknown")
+
+                    # Handle image tag from different possible locations
+                    image_name = result.get("tag", image.get("tag", "unknown:latest"))
+
+                    trigger_id = result.get("triggerId", "unknown")
+
+                    # Split repo and tag safely
+                    if ":" in image_name:
+                        repo, tag = image_name.split(":", 1)
+                    else:
+                        repo = image_name
+                        tag = "latest"
+
                     severity, active = get_severity(status, description)
                     vulnerability_id = extract_vulnerability_id(trigger_id)
-                    title = (
-                        policy_id
-                        + " - gate|"
-                        + gate
-                        + " - trigger|"
-                        + trigger_id
-                    )
+                    title = policy_id + " - gate|" + gate + " - trigger|" + trigger_id
                     find = Finding(
                         title=title,
                         test=test,
@@ -74,8 +122,10 @@ def get_findings(self, filename, test):
                         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)
+                    msg = f"Invalid format or missing key: {err}. This parser supports both legacy AnchoreCTL format and the new format from 'anchorectl policy evaluate -o json'."
+                    logger.warning(msg)
+                    # Continue processing other findings instead of failing completely
+                    continue
         return items
 
 

unittests/scans/anchorectl_policies/new_format_many_violations.json

@@ -0,0 +1,46 @@
+{
+  "evaluatedTag": "test/testimage:testtag",
+  "policyId": "9e104ade-7b57-4cdc-93fb-4949bf3b36b6",
+  "imageDigest": "sha256:8htz0bf942cfcd6hg8cf6435afd318b65d23e4c1a80044304c6e3ed20",
+  "evaluations": [
+    {
+      "evaluationTime": "2022-09-20T08:25:52Z",
+      "status": "fail",
+      "finalAction": "stop",
+      "finalActionReason": "policy_evaluation",
+      "numberOfFindings": 3,
+      "details": [
+        {
+          "description": "HIGH Vulnerability found in non-os package type (test) - /usr/local/bin/testbinary (CVE-2022-1234)",
+          "gate": "vulnerabilities",
+          "imageId": "d26f0119b9634091a541b081dd8bdca435ab52e114e4b4328575c0bc2c69768b",
+          "policyId": "SoftwareChecks",
+          "status": "stop",
+          "tag": "test/testimage:testtag",
+          "triggerId": "CVE-2022-1234+test",
+          "triggerName": "package"
+        },
+        {
+          "description": "MEDIUM Vulnerability found in non-os package type (test2) - /usr/local/bin/testbinary (fixed in: 1.2.3)(GHSA-1234-abcd-5678)",
+          "gate": "vulnerabilities",
+          "imageId": "d26f0119b9634091a541b081dd8bdca435ab52e114e4b4328575c0bc2c69768b",
+          "policyId": "SoftwareChecks",
+          "status": "stop",
+          "tag": "test/testimage:testtag",
+          "triggerId": "GHSA-1234-abcd-5678+test2",
+          "triggerName": "package"
+        },
+        {
+          "description": "User root found as effective user, which is not on the allowed list",
+          "gate": "dockerfile",
+          "imageId": "d26f0119b9634091a541b081dd8bdca435ab52e114e4b4328575c0bc2c69768b",
+          "policyId": "RootUser",
+          "status": "warn",
+          "tag": "test/testimage:testtag",
+          "triggerId": "b2605c2ddbdb02b8e2365c9248dada5a",
+          "triggerName": "effective_user"
+        }
+      ]
+    }
+  ]
+}

unittests/scans/anchorectl_policies/new_format_no_violation.json

@@ -0,0 +1,15 @@
+{
+  "evaluatedTag": "test/testimage:testtag",
+  "policyId": "9e104ade-7b57-4cdc-93fb-4949bf3b36b6",
+  "imageDigest": "sha256:8htz0bf942cfcd6hg8cf6435afd318b65d23e4c1a80044304c6e3ed20",
+  "evaluations": [
+    {
+      "evaluationTime": "2022-09-20T08:25:52Z",
+      "status": "pass",
+      "finalAction": "go",
+      "finalActionReason": "policy_evaluation",
+      "numberOfFindings": 0,
+      "details": []
+    }
+  ]
+}

unittests/scans/anchorectl_policies/new_format_one_violation.json

@@ -0,0 +1,26 @@
+{
+  "evaluatedTag": "test/testimage:testtag",
+  "policyId": "9e104ade-7b57-4cdc-93fb-4949bf3b36b6",
+  "imageDigest": "sha256:8htz0bf942cfcd6hg8cf6435afd318b65d23e4c1a80044304c6e3ed20",
+  "evaluations": [
+    {
+      "evaluationTime": "2022-09-20T08:25:52Z",
+      "status": "fail",
+      "finalAction": "stop",
+      "finalActionReason": "policy_evaluation",
+      "numberOfFindings": 1,
+      "details": [
+        {
+          "description": "User root found as effective user, which is not on the allowed list",
+          "gate": "dockerfile",
+          "imageId": "d26f0119b9634091a541b081dd8bdca435ab52e114e4b4328575c0bc2c69768b",
+          "policyId": "RootUser",
+          "status": "warn",
+          "tag": "test/testimage:testtag",
+          "triggerId": "b2605c2ddbdb02b8e2365c9248dada5a",
+          "triggerName": "effective_user"
+        }
+      ]
+    }
+  ]
+}

unittests/scans/anchorectl_policies/new_format_one_violation_description_severity.json

@@ -0,0 +1,26 @@
+{
+  "evaluatedTag": "test/testimage:testtag",
+  "policyId": "9e104ade-7b57-4cdc-93fb-4949bf3b36b6",
+  "imageDigest": "sha256:8htz0bf942cfcd6hg8cf6435afd318b65d23e4c1a80044304c6e3ed20",
+  "evaluations": [
+    {
+      "evaluationTime": "2022-09-20T08:25:52Z",
+      "status": "fail",
+      "finalAction": "stop",
+      "finalActionReason": "policy_evaluation",
+      "numberOfFindings": 1,
+      "details": [
+        {
+          "description": "CRITICAL User root found as effective user, which is not on the allowed list",
+          "gate": "dockerfile",
+          "imageId": "d26f0119b9634091a541b081dd8bdca435ab52e114e4b4328575c0bc2c69768b",
+          "policyId": "RootUser",
+          "status": "warn",
+          "tag": "test/testimage:testtag",
+          "triggerId": "b2605c2ddbdb02b8e2365c9248dada5a",
+          "triggerName": "effective_user"
+        }
+      ]
+    }
+  ]
+}

unittests/tools/test_anchorectl_policies_parser.py

@@ -35,3 +35,36 @@ def test_anchore_engine_parser_has_one_finding_and_description_has_severity(self
             self.assertEqual(singleFinding.severity, "Critical")
             self.assertEqual(singleFinding.title, "RootUser - gate|dockerfile - trigger|b2605c2ddbdb02b8e2365c9248dada5a")
             self.assertEqual(singleFinding.description, "CRITICAL User root found as effective user, which is not on the allowed list")
+
+    # Tests for the new AnchoreCTL format
+    def test_new_format_anchore_engine_parser_has_no_finding(self):
+        with (get_unit_tests_scans_path("anchorectl_policies") / "new_format_no_violation.json").open(encoding="utf-8") as testfile:
+            parser = AnchoreCTLPoliciesParser()
+            findings = parser.get_findings(testfile, Test())
+            self.assertEqual(0, len(findings))
+
+    def test_new_format_anchore_engine_parser_has_one_finding_and_it_is_correctly_parsed(self):
+        with (get_unit_tests_scans_path("anchorectl_policies") / "new_format_one_violation.json").open(encoding="utf-8") as testfile:
+            parser = AnchoreCTLPoliciesParser()
+            findings = parser.get_findings(testfile, Test())
+            self.assertEqual(1, len(findings))
+            singleFinding = findings[0]
+            self.assertEqual(singleFinding.severity, "Medium")
+            self.assertEqual(singleFinding.title, "RootUser - gate|dockerfile - trigger|b2605c2ddbdb02b8e2365c9248dada5a")
+            self.assertEqual(singleFinding.description, "User root found as effective user, which is not on the allowed list")
+
+    def test_new_format_anchore_engine_parser_has_many_findings(self):
+        with (get_unit_tests_scans_path("anchorectl_policies") / "new_format_many_violations.json").open(encoding="utf-8") as testfile:
+            parser = AnchoreCTLPoliciesParser()
+            findings = parser.get_findings(testfile, Test())
+            self.assertEqual(3, len(findings))
+
+    def test_new_format_anchore_engine_parser_has_one_finding_and_description_has_severity(self):
+        with (get_unit_tests_scans_path("anchorectl_policies") / "new_format_one_violation_description_severity.json").open(encoding="utf-8") as testfile:
+            parser = AnchoreCTLPoliciesParser()
+            findings = parser.get_findings(testfile, Test())
+            self.assertEqual(1, len(findings))
+            singleFinding = findings[0]
+            self.assertEqual(singleFinding.severity, "Critical")
+            self.assertEqual(singleFinding.title, "RootUser - gate|dockerfile - trigger|b2605c2ddbdb02b8e2365c9248dada5a")
+            self.assertEqual(singleFinding.description, "CRITICAL User root found as effective user, which is not on the allowed list")