Improve tag handling in DefaultImporter and add tests for tag imports

dojo/importers/default_importer.py

@@ -226,7 +226,11 @@ def process_findings(
             # Process any endpoints on the endpoint, or added on the form
             self.process_endpoints(finding, self.endpoints_to_add)
             # Parsers must use unsaved_tags to store tags, so we can clean them
-            finding.tags = clean_tags(finding.unsaved_tags)
+            cleaned_tags = clean_tags(finding.unsaved_tags)
+            if isinstance(cleaned_tags, list):
+                finding.tags.set(cleaned_tags)
+            elif isinstance(cleaned_tags, str):
+                finding.tags.set([cleaned_tags])
             # Process any files
             self.process_files(finding)
             # Process vulnerability IDs

dojo/importers/default_reimporter.py

@@ -693,8 +693,12 @@ def finding_post_processing(
         if len(self.endpoints_to_add) > 0:
             self.endpoint_manager.chunk_endpoints_and_disperse(finding, self.endpoints_to_add)
         # Parsers must use unsaved_tags to store tags, so we can clean them
-        if finding.unsaved_tags:
-            finding.tags = clean_tags(finding.unsaved_tags)
+        if finding_from_report.unsaved_tags:
+            cleaned_tags = clean_tags(finding_from_report.unsaved_tags)
+            if isinstance(cleaned_tags, list):
+                finding.tags.set(cleaned_tags)
+            elif isinstance(cleaned_tags, str):
+                finding.tags.set([cleaned_tags])
         # Process any files
         if finding_from_report.unsaved_files:
             finding.unsaved_files = finding_from_report.unsaved_files

unittests/scans/generic/generic_report1_more_tags.json

@@ -0,0 +1,34 @@
+{
+    "findings": [
+        {
+            "title": "test title",
+            "description": "Some very long description with\n\n some UTF-8 chars à qu'il est beau",
+            "active": true,
+            "verified": true,
+            "severity": "Medium",
+            "impact": "Some impact",
+            "date": "2021-01-06",
+            "cve": "CVE-2020-36234",
+            "cwe": 261,
+            "cvssv3": "CVSS:3.1/AV:N/AC:L/PR:H/UI:R/S:C/C:L/I:L/A:N",
+            "tags": ["security", "network", "hardened"],
+            "unique_id_from_tool": "3287f2d0-554f-491b-8516-3c349ead8ee5",
+            "vuln_id_from_tool": "TEST1"
+        },
+        {
+            "title": "test title2",
+            "description": "Some very long description with\n\n some UTF-8 chars à qu'il est beau2",
+            "active": true,
+            "verified": false,
+            "severity": "Medium",
+            "impact": "Some impact",
+            "date": "2021-01-06",
+            "cve": "CVE-2020-36235",
+            "cwe": 287,
+            "cvssv3": "CVSS:3.1/AV:N/AC:L/PR:H/UI:R/S:C/C:L/I:L/A:N",
+            "tags": ["security", "network", "hardened"],
+            "unique_id_from_tool": "42500af3-68c5-4dc3-8022-191d93c2f1f7",
+            "vuln_id_from_tool": "TEST2"
+        }
+    ]
+}

unittests/test_tags.py

@@ -15,8 +15,9 @@ class TagTests(DojoAPITestCase):
     def setUp(self, *args, **kwargs):
         super().setUp()
         self.login_as_admin()
-        self.scans_path = get_unit_tests_scans_path("zap")
-        self.zap_sample5_filename = self.scans_path / "5_zap_sample_one.xml"
+        self.zap_sample5_filename = get_unit_tests_scans_path("zap") / "5_zap_sample_one.xml"
+        self.generic_sample_with_tags_filename = get_unit_tests_scans_path("generic") / "generic_report1.json"
+        self.generic_sample_with_more_tags_filename = get_unit_tests_scans_path("generic") / "generic_report1_more_tags.json"
 
     def test_create_product_with_tags(self, expected_status_code: int = 201):
         product_id = Product.objects.all().first().id
@@ -285,6 +286,28 @@ def test_import_multipart_tags(self):
             for tag in success_tags:
                 self.assertIn(tag, response["tags"])
 
+    def test_import_report_with_tags(self):
+        def assert_tags_in_findings(findings: list[dict], expected_finding_count: int, desired_tags: list[str]) -> None:
+            self.assertEqual(expected_finding_count, len(findings))
+            for finding in findings:
+                self.assertEqual(len(desired_tags), len(finding.get("tags")))
+                for tag in desired_tags:
+                    self.assertIn(tag, finding["tags"])
+
+        # Import a report with findings that have tags
+        import0 = self.import_scan_with_params(self.generic_sample_with_tags_filename, scan_type="Generic Findings Import")
+        test_id = import0["test"]
+        response = self.get_test_findings_api(test_id)
+        findings = response["results"]
+        # Make sure we have what we are looking for
+        assert_tags_in_findings(findings, 2, ["security", "network"])
+        # Reimport with a different report that has more tags
+        self.reimport_scan_with_params(test_id, self.generic_sample_with_more_tags_filename, scan_type="Generic Findings Import")
+        response = self.get_test_findings_api(test_id)
+        findings = response["results"]
+        # Make sure we have what we are looking for
+        assert_tags_in_findings(findings, 2, ["security", "network", "hardened"])
+
 
 class InheritedTagsTests(DojoAPITestCase):
     fixtures = ["dojo_testdata.json"]