Add scanner name to Test name

Author: mfyll

dojo/tools/openreports/parser.py

@@ -4,6 +4,7 @@
 import logging
 
 from dojo.models import Finding
+from dojo.tools.parser_test import ParserTest
 
 logger = logging.getLogger(__name__)
 
@@ -72,6 +73,61 @@ def get_findings(self, scan_file, test):
 
         return findings
 
+    def get_tests(self, scan_type, handle):
+        try:
+            data = json.load(handle)
+        except Exception:
+            handle.seek(0)
+            scan_data = handle.read()
+            try:
+                data = json.loads(str(scan_data, "utf-8"))
+            except Exception:
+                data = json.loads(scan_data)
+
+        if data is None:
+            return []
+
+        # Handle both single report and list of reports
+        reports = []
+        if isinstance(data, dict):
+            if data.get("kind") == "List" and "items" in data:
+                reports = data["items"]
+            elif data.get("kind") == "Report":
+                reports = [data]
+        elif isinstance(data, list):
+            reports = data
+
+        # Find all unique sources across all reports
+        sources_found = set()
+        for report in reports:
+            if not isinstance(report, dict) or report.get("kind") != "Report":
+                continue
+            for result in report.get("results", []):
+                source = result.get("source", "OpenReports")
+                sources_found.add(source)
+
+        # Create a ParserTest for each source
+        tests = []
+        for source in sorted(sources_found):
+            test = ParserTest(
+                name=source,
+                parser_type=source,
+                version=None,
+            )
+            test.findings = []
+
+            # Parse all reports and filter findings by source
+            for report in reports:
+                if not isinstance(report, dict) or report.get("kind") != "Report":
+                    continue
+
+                findings = self._parse_report_for_source(test, report, source)
+                test.findings.extend(findings)
+
+            tests.append(test)
+
+        return tests
+
     def _parse_report(self, test, report):
         findings = []
 
@@ -102,6 +158,41 @@ def _parse_report(self, test, report):
 
         return findings
 
+    def _parse_report_for_source(self, test, report, source_filter):
+        findings = []
+
+        # Extract metadata
+        metadata = report.get("metadata", {})
+        report_name = metadata.get("name", "")
+        namespace = metadata.get("namespace", "")
+        report_uid = metadata.get("uid", "")
+
+        # Extract scope information
+        scope = report.get("scope", {})
+        scope_kind = scope.get("kind", "")
+        scope_name = scope.get("name", "")
+
+        # Create service identifier from scope and metadata
+        service_name = f"{namespace}/{scope_kind}/{scope_name}" if namespace else f"{scope_kind}/{scope_name}"
+
+        # Extract results
+        results = report.get("results", [])
+
+        for result in results:
+            if not isinstance(result, dict):
+                continue
+
+            # Filter by source
+            result_source = result.get("source", "OpenReports")
+            if result_source != source_filter:
+                continue
+
+            finding = self._create_finding_from_result(None, result, service_name, report_name, report_uid)
+            if finding:
+                findings.append(finding)
+
+        return findings
+
     def _create_finding_from_result(self, test, result, service_name, report_name, report_uid):
         try:
             # Extract basic fields

unittests/tools/test_openreports_parser.py

@@ -35,7 +35,8 @@ def test_single_report(self):
             self.assertEqual(1, len(finding1.unsaved_vulnerability_ids))
             self.assertEqual("CVE-2025-9232", finding1.unsaved_vulnerability_ids[0])
             self.assertEqual(
-                "b1fcca57-2efd-44d3-89e9-949e29b61936:CVE-2025-9232:libcrypto3", finding1.unique_id_from_tool,
+                "b1fcca57-2efd-44d3-89e9-949e29b61936:CVE-2025-9232:libcrypto3",
+                finding1.unique_id_from_tool,
             )
             self.assertIn("vulnerability scan", finding1.tags)
             self.assertIn("image-scanner", finding1.tags)
@@ -72,7 +73,8 @@ def test_single_report(self):
             # Non-CVE policies should not have vulnerability IDs
             self.assertIsNone(finding3.unsaved_vulnerability_ids)
             self.assertEqual(
-                "b1fcca57-2efd-44d3-89e9-949e29b61936:CIS-BENCH-001:web-server", finding3.unique_id_from_tool,
+                "b1fcca57-2efd-44d3-89e9-949e29b61936:CIS-BENCH-001:web-server",
+                finding3.unique_id_from_tool,
             )
             self.assertIn("compliance check", finding3.tags)
             self.assertIn("compliance-scanner", finding3.tags)
@@ -113,3 +115,56 @@ def test_parser_metadata(self):
 
         description = parser.get_description_for_scan_types("OpenReports")
         self.assertEqual("Import OpenReports JSON report.", description)
+
+    def test_get_tests_single_source(self):
+        with sample_path("openreports_single_report.json").open(encoding="utf-8") as test_file:
+            parser = OpenreportsParser()
+            tests = parser.get_tests("OpenReports", test_file)
+
+            # Should have two tests for the two sources
+            self.assertEqual(len(tests), 2)
+
+            # Verify test names
+            test_names = {test.name for test in tests}
+            self.assertIn("image-scanner", test_names)
+            self.assertIn("compliance-scanner", test_names)
+
+            # Find the image-scanner test
+            image_scanner_test = next(t for t in tests if t.name == "image-scanner")
+            self.assertEqual("image-scanner", image_scanner_test.type)
+            self.assertIsNone(image_scanner_test.version)
+            self.assertEqual(2, len(image_scanner_test.findings))
+
+            # Verify findings are properly created
+            finding1 = image_scanner_test.findings[0]
+            self.assertEqual("CVE-2025-9232 in libcrypto3", finding1.title)
+            self.assertEqual("Low", finding1.severity)
+            # Verify test is not set - check using hasattr to avoid RelatedObjectDoesNotExist
+            self.assertFalse(hasattr(finding1, "test") and finding1.test is not None)
+
+    def test_get_tests_multiple_sources(self):
+        with sample_path("openreports_list_format.json").open(encoding="utf-8") as test_file:
+            parser = OpenreportsParser()
+            tests = parser.get_tests("OpenReports", test_file)
+
+            # Should have two tests for the two different sources
+            self.assertEqual(len(tests), 2)
+
+            # Verify test names
+            test_names = {test.name for test in tests}
+            self.assertIn("policy-scanner", test_names)
+            self.assertIn("image-scanner", test_names)
+
+            # Find the image-scanner test
+            image_scanner_test = next(t for t in tests if t.name == "image-scanner")
+            self.assertEqual(2, len(image_scanner_test.findings))
+
+            # Find the policy-scanner test
+            policy_scanner_test = next(t for t in tests if t.name == "policy-scanner")
+            self.assertEqual(1, len(policy_scanner_test.findings))
+
+            # Verify findings have no test set
+            for test in tests:
+                for finding in test.findings:
+                    # Check using hasattr to avoid RelatedObjectDoesNotExist
+                    self.assertFalse(hasattr(finding, "test") and finding.test is not None)