🔨 Rework RustyHog to fix #10584

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" or directly the specific sub scanner (e.g. "Duroc Hog Scan"). If you choose "Rusty Hog Scan", we reommend to reupload 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)
                 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)
                 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)
                 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)
                 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)