🔨 RustyHog: handle empty reports correctly to fix #10584 (#12129)

* 🔨 Rework RustyHog to fix #10584

* Update docs/content/en/connecting_your_tools/parsers/file/rusty_hog.md

Co-authored-by: Cody Maffucci <46459665+Maffooch@users.noreply.github.com>

* update

---------

Co-authored-by: Cody Maffucci <46459665+Maffooch@users.noreply.github.com>

Author: manuel-sommer

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

@@ -14,5 +14,7 @@ DefectDojo currently supports the parsing of the following Rusty Hog JSON output
 RustyHog scans only one target at a time. This is not efficient if you want to scan all targets (e.g. all JIRA tickets) and upload each single report to DefectDojo.
 [Rusty-Hog-Wrapper](https://github.com/manuel-sommer/Rusty-Hog-Wrapper) deals with this and scans a whole JIRA Project or Confluence Space, merges the findings into a valid file which can be uploaded to DefectDojo. (This is no official recommendation from DefectDojo, but rather a pointer in a direction on how to use this vulnerability scanner in a more efficient way.)
 
+You can either select "Rusty Hog Scan" directly, or specify the sub scanner (e.g. "Duroc Hog Scan"). If you choose "Rusty Hog Scan", we recommend to re-import scans into the same test. For more information look at [this issue](https://github.com/DefectDojo/django-DefectDojo/issues/10584).
+
 ### Sample Scan Data
 Sample Rusty Hog parser scans can be found [here](https://github.com/DefectDojo/django-DefectDojo/tree/master/unittests/scans/rusty_hog).

dojo/tools/rusty_hog/parser.py

@@ -6,68 +6,66 @@
 
 class RustyhogParser:
     def get_scan_types(self):
-        return ["Rusty Hog Scan"]
+        return ["Rusty Hog Scan", "Choctaw Hog Scan", "Duroc Hog Scan", "Gottingen Hog Scan", "Essex Hog Scan"]
 
     def get_label_for_scan_types(self, scan_type):
         return scan_type  # no custom label for now
 
     def get_description_for_scan_types(self, scan_type):
         return "Rusty Hog Scan - JSON Report"
 
-    def get_findings(self, json_output, test):
-        tree = json.load(json_output)
-        return self.get_items(tree, test)
-
-    def parse_json(self, json_output):
-        return json.load(json_output)
-
-    def get_items(self, json_output, scanner, test):
+    def get_findings(self, filename, test):
+        if filename:
+            tree = filename.read()
+            try:
+                json_output = json.loads(str(tree, "utf-8"))
+            except Exception:
+                json_output = json.loads(tree)
+            if not json_output:
+                json_output = []
         items = {}
-        findings = self.__getitem(
-            vulnerabilities=self.parse_json(json_output), scanner=scanner,
-        )
+        findings = self.__getitem(vulnerabilities=json_output, scanner=test)
         for finding in findings:
             unique_key = f"Finding {finding}"
             items[unique_key] = finding
         return list(items.values())
 
     def get_tests(self, scan_type, handle):
-        tree = self.parse_json(handle)
+        tree = json.load(handle)
         tests = []
-        parsername = "Rusty Hog"
-        for node in tree:
-            if (
-                "commit" in node
-                or "commitHash" in node
-                or "parent_commit_hash" in node
-                or "old_file_id" in node
-                or "new_file_id" in node
-            ):
-                parsername = "Choctaw Hog"
-                break
-            if "linenum" in node or "diff" in node:
-                parsername = "Duroc Hog"
-                break
-            if "issue_id" in node or "location" in node:
-                parsername = "Gottingen Hog"
-                break
-            if "page_id" in node:
-                parsername = "Essex Hog"
-                break
+        if scan_type == "Rusty Hog Scan":
+            parsername = "Rusty Hog"
+            for node in tree:
+                if (
+                    "commit" in node
+                    or "commitHash" in node
+                    or "parent_commit_hash" in node
+                    or "old_file_id" in node
+                    or "new_file_id" in node
+                ):
+                    parsername = "Choctaw Hog"
+                    break
+                if "linenum" in node or "diff" in node:
+                    parsername = "Duroc Hog"
+                    break
+                if "issue_id" in node or "location" in node:
+                    parsername = "Gottingen Hog"
+                    break
+                if "page_id" in node:
+                    parsername = "Essex Hog"
+                    break
+        else:
+            parsername = scan_type.replace(" Scan", "")
         test = ParserTest(
             name=parsername,
             parser_type=parsername,
             version="",
         )
-        if (
-            parsername == "Rusty Hog"
-        ):  # The outputfile is empty. A subscanner can't be classified
+        if parsername == "Rusty Hog":  # The outputfile is empty. A subscanner can't be classified
             test.description = "The exact scanner within Rusty Hog could not be determined due to missing information within the scan result."
         else:
             test.description = parsername
-        test.findings = self.__getitem(
-            vulnerabilities=tree, scanner=parsername,
-        )
+        test.findings = self.__getitem(vulnerabilities=tree, scanner=parsername)
         tests.append(test)
         return tests
 
@@ -85,13 +83,15 @@ def __getitem(self, vulnerabilities, scanner):
                 break
             if scanner == "Choctaw Hog":
                 """Choctaw Hog"""
+                if vulnerability.get("commitHash") is None:
+                    raise ValueError("You chose the wrong scan type: " + scanner + ". A commitHash is expected.")
                 found_secret_string = str(vulnerability.get("stringsFound") or "")
-                description += f"**This string was found:** {found_secret_string}"
+                description += f"\n**This string was found:** {found_secret_string}"
                 if vulnerability.get("commit") is not None:
                     description += "\n**Commit message:** {}".format(
                         vulnerability.get("commit"),
                     )
-                if vulnerability.get("commitHash") is not None:
+                if vulnerability.get("commitHash"):
                     description += "\n**Commit hash:** {}".format(
                         vulnerability.get("commitHash"),
                     )
@@ -121,13 +121,15 @@ def __getitem(self, vulnerabilities, scanner):
                     )
             elif scanner == "Duroc Hog":
                 """Duroc Hog"""
+                if vulnerability.get("linenum") is None:
+                    raise ValueError("You chose the wrong scan type: " + scanner + ". A linenum is expected.")
                 found_secret_string = str(vulnerability.get("stringsFound") or "")
-                description += f"**This string was found:** {found_secret_string}"
+                description += f"\n**This string was found:** {found_secret_string}"
                 if vulnerability.get("path") is not None:
                     description += "\n**Path of Issue:** {}".format(
                         vulnerability.get("path"),
                     )
-                if vulnerability.get("linenum") is not None:
+                if vulnerability.get("linenum"):
                     description += "\n**Linenum of Issue:** {}".format(
                         vulnerability.get("linenum"),
                     )
@@ -137,9 +139,11 @@ def __getitem(self, vulnerabilities, scanner):
                     )
             elif scanner == "Gottingen Hog":
                 """Gottingen Hog"""
+                if vulnerability.get("issue_id") is None:
+                    raise ValueError("You chose the wrong scan type: " + scanner + ". An issue_id is expected.")
                 found_secret_string = str(vulnerability.get("stringsFound") or "")
-                description += f"**This string was found:** {found_secret_string}"
-                if vulnerability.get("issue_id") is not None:
+                description += f"\n**This string was found:** {found_secret_string}"
+                if vulnerability.get("issue_id"):
                     description += "\n**JIRA Issue ID:** {}".format(
                         vulnerability.get("issue_id"),
                     )
@@ -152,9 +156,11 @@ def __getitem(self, vulnerabilities, scanner):
                         vulnerability.get("url"), vulnerability.get("url"),
                     )
             elif scanner == "Essex Hog":
+                if vulnerability.get("page_id") is None:
+                    raise ValueError("You chose the wrong scan type: " + scanner + ". A page_id is expected.")
                 found_secret_string = str(vulnerability.get("stringsFound") or "")
-                description += f"**This string was found:** {found_secret_string}"
-                if vulnerability.get("page_id") is not None:
+                description += f"\n**This string was found:** {found_secret_string}"
+                if vulnerability.get("page_id"):
                     description += "\n**Confluence URL:** [{}]({})".format(
                         vulnerability.get("url"), vulnerability.get("url"),
                     )

unittests/tools/test_rusty_hog_parser.py

@@ -1,4 +1,3 @@
-from dojo.models import Test
 from dojo.tools.rusty_hog.parser import RustyhogParser
 from unittests.dojo_test_case import DojoTestCase, get_unit_tests_scans_path
 
@@ -7,25 +6,25 @@ class TestRustyhogParser(DojoTestCase):
     def test_parse_file_with_no_vuln_has_no_finding_choctawhog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "choctawhog_no_vuln.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Rusty Hog", Test())  # The outputfile is empty. A subscanner can't be classified
+            findings = parser.get_findings(testfile, "Rusty Hog")  # The outputfile is empty. A subscanner can't be classified
             self.assertEqual(0, len(findings))
 
     def test_parse_file_with_one_vuln_has_one_finding_choctawhog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "choctawhog_one_vuln.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Choctaw Hog", Test())
+            findings = parser.get_findings(testfile, "Choctaw Hog")
             self.assertEqual(1, len(findings))
 
     def test_parse_file_with_multiple_vuln_has_multiple_finding_choctawhog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "choctawhog_many_vulns.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Choctaw Hog", Test())
+            findings = parser.get_findings(testfile, "Choctaw Hog")
             self.assertEqual(13, len(findings))
 
     def test_parse_file_with_multiple_vuln_has_multiple_finding_choctawhog_content(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "choctawhog_many_vulns.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Choctaw Hog", Test())
+            findings = parser.get_findings(testfile, "Choctaw Hog")
             self.assertEqual(findings[0].title, "Email address found in Git path .github/workflows/main.yml (a7bce96377c4ff2ac16cd51fb0da7fe7ea678829)")
             self.assertIn("**This string was found:** ['dojo-helpers@this-repo.com']", findings[0].description)
             self.assertIn("**Commit message:** removing action", findings[0].description)
@@ -38,25 +37,25 @@ def test_parse_file_with_multiple_vuln_has_multiple_finding_choctawhog_content(s
     def test_parse_file_with_no_vuln_has_no_finding_duorchog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "durochog_no_vuln.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Rusty Hog", Test())  # The outputfile is empty. A subscanner can't be classified
+            findings = parser.get_findings(testfile, "Rusty Hog")  # The outputfile is empty. A subscanner can't be classified
             self.assertEqual(0, len(findings))
 
     def test_parse_file_with_one_vuln_has_one_finding_durochog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "durochog_one_vuln.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Duroc Hog", Test())
+            findings = parser.get_findings(testfile, "Duroc Hog")
             self.assertEqual(1, len(findings))
 
     def test_parse_file_with_multiple_vuln_has_multiple_finding_durochog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "durochog_many_vulns.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Duroc Hog", Test())
+            findings = parser.get_findings(testfile, "Duroc Hog")
             self.assertEqual(4, len(findings))
 
     def test_parse_file_with_multiple_vuln_has_multiple_finding_durochog_content(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "durochog_many_vulns.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Duroc Hog", Test())
+            findings = parser.get_findings(testfile, "Duroc Hog")
             self.assertEqual(findings[0].title, "password (Password) found in path /scan_folder/unittests/scans/sonarqube/sonar-no-finding.html")
             self.assertIn("**This string was found:** ['password = getEncryptedPass()']", findings[0].description)
             self.assertIn("**Path of Issue:** /scan_folder/unittests/scans/sonarqube/sonar-no-finding.html", findings[0].description)
@@ -67,25 +66,25 @@ def test_parse_file_with_multiple_vuln_has_multiple_finding_durochog_content(sel
     def test_parse_file_with_no_vuln_has_no_finding_gottingenhog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "gottingenhog_no_vuln.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Rusty Hog", Test())  # The outputfile is empty. A subscanner can't be classified
+            findings = parser.get_findings(testfile, "Rusty Hog")  # The outputfile is empty. A subscanner can't be classified
             self.assertEqual(0, len(findings))
 
     def test_parse_file_with_one_vuln_has_one_finding_gottingenhog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "gottingenhog_one_vuln.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Gottingen Hog", Test())
+            findings = parser.get_findings(testfile, "Gottingen Hog")
             self.assertEqual(1, len(findings))
 
     def test_parse_file_with_multiple_vuln_has_multiple_finding_gottingenhog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "gottingenhog_many_vulns.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Gottingen Hog", Test())
+            findings = parser.get_findings(testfile, "Gottingen Hog")
             self.assertEqual(10, len(findings))
 
     def test_parse_file_with_multiple_vuln_has_multiple_finding_gottingenhog_content(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "gottingenhog_many_vulns.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Gottingen Hog", Test())
+            findings = parser.get_findings(testfile, "Gottingen Hog")
             self.assertEqual(findings[0].title, "password found in Jira ID TEST-123 (Issue Description)")
             self.assertIn("**This string was found:** ['password: jeans']", findings[0].description)
             self.assertIn("**JIRA Issue ID:** TEST-123", findings[0].description)
@@ -96,19 +95,19 @@ def test_parse_file_with_multiple_vuln_has_multiple_finding_gottingenhog_content
     def test_parse_file_with_no_vuln_has_no_finding_essexhog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "essexhog_no_vuln.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Rusty Hog", Test())  # The outputfile is empty. A subscanner can't be classified
+            findings = parser.get_findings(testfile, "Rusty Hog")  # The outputfile is empty. A subscanner can't be classified
             self.assertEqual(0, len(findings))
 
     def test_parse_file_with_one_vuln_has_one_finding_essexhog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "essexhog_one_vuln.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Essex Hog", Test())
+            findings = parser.get_findings(testfile, "Essex Hog")
             self.assertEqual(1, len(findings))
 
     def test_parse_file_with_multiple_vuln_has_multiple_finding_essexhog(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "essexhog_many_vulns.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Essex Hog", Test())
+            findings = parser.get_findings(testfile, "Essex Hog")
             self.assertEqual(3, len(findings))
             self.assertEqual("https://confluence.com/pages/viewpage.action?pageId=12345", findings[0].file_path)
             self.assertEqual("['-----BEGIN EC PRIVATE KEY-----']", findings[0].payload)
@@ -117,7 +116,7 @@ def test_parse_file_with_multiple_vuln_has_multiple_finding_essexhog(self):
     def test_parse_file_with_multiple_vuln_has_multiple_finding_essexhog_content(self):
         with open(get_unit_tests_scans_path("rusty_hog") / "essexhog_many_vulns.json", encoding="utf-8") as testfile:
             parser = RustyhogParser()
-            findings = parser.get_items(testfile, "Essex Hog", Test())
+            findings = parser.get_findings(testfile, "Essex Hog")
             self.assertEqual(findings[0].title, "SSH (EC) private key found in Confluence Page ID 12345")
             self.assertIn("-----BEGIN EC PRIVATE KEY-----", findings[0].description)
             self.assertIn("**Confluence URL:** [https://confluence.com/pages/viewpage.action?pageId=12345](https://confluence.com/pages/viewpage.action?pageId=12345)", findings[0].description)