Updating pip-audit parser to handle new JSON file format

docs/content/en/integrations/parsers/file/pip_audit.md

@@ -2,7 +2,41 @@
 title: "pip-audit Scan"
 toc_hide: true
 ---
-Import pip-audit JSON scan report
+
+Import pip-audit JSON scan report.
+
+### File Types
+This parser expects a JSON file.
+
+The parser can handle legacy and current JSON format.
+
+The current format has added a `dependencies` element:
+
+    {
+	  "dependencies": [
+	    {
+	      "name": "pyopenssl",
+	      "version": "23.1.0",
+	      "vulns": []
+	    },
+	...
+	  ]
+	...
+	}
+
+The legacy format does not include the `dependencies` key:
+
+    [
+	    {
+	        "name": "adal",
+	        "version": "1.2.2",
+	        "vulns": []
+	    },
+    ...
+    ]
 
 ### Sample Scan Data
-Sample pip-audit Scan scans can be found [here](https://github.com/DefectDojo/django-DefectDojo/tree/master/unittests/scans/pip_audit).
+Sample pip-audit Scan scans can be found [here](https://github.com/DefectDojo/django-DefectDojo/tree/master/unittests/scans/pip_audit).
+
+### Link To Tool
+[pip-audit](https://pypi.org/project/pip-audit/)

dojo/tools/pip_audit/parser.py

@@ -1,70 +1,110 @@
+"""Parser for pip-audit."""
 import json
 
 from dojo.models import Finding
 
 
 class PipAuditParser:
+    """Represents a file parser capable of ingesting pip-audit results."""
+
     def get_scan_types(self):
+        """Return the type of scan this parser ingests."""
         return ["pip-audit Scan"]
 
     def get_label_for_scan_types(self, scan_type):
+        """Return the friendly name for this parser."""
         return "pip-audit Scan"
 
     def get_description_for_scan_types(self, scan_type):
+        """Return the description for this parser."""
         return "Import pip-audit JSON scan report."
 
     def requires_file(self, scan_type):
+        """Return boolean indicating if parser requires a file to process."""
         return True
 
     def get_findings(self, scan_file, test):
+        """Return the collection of Findings ingested."""
         data = json.load(scan_file)
-
-        findings = list()
-        for item in data:
-            vulnerabilities = item.get("vulns", [])
-            if vulnerabilities:
-                component_name = item["name"]
-                component_version = item.get("version")
-                for vulnerability in vulnerabilities:
-                    vuln_id = vulnerability.get("id")
-                    vuln_fix_versions = vulnerability.get("fix_versions")
-                    vuln_description = vulnerability.get("description")
-
-                    title = (
-                        f"{vuln_id} in {component_name}:{component_version}"
-                    )
-
-                    description = ""
-                    description += vuln_description
-
-                    mitigation = None
-                    if vuln_fix_versions:
-                        mitigation = "Upgrade to version:"
-                        if len(vuln_fix_versions) == 1:
-                            mitigation += f" {vuln_fix_versions[0]}"
-                        else:
-                            for fix_version in vuln_fix_versions:
-                                mitigation += f"\n- {fix_version}"
-
-                    finding = Finding(
-                        test=test,
-                        title=title,
-                        cwe=1352,
-                        severity="Medium",
-                        description=description,
-                        mitigation=mitigation,
-                        component_name=component_name,
-                        component_version=component_version,
-                        vuln_id_from_tool=vuln_id,
-                        static_finding=True,
-                        dynamic_finding=False,
-                    )
-                    vulnerability_ids = list()
-                    if vuln_id:
-                        vulnerability_ids.append(vuln_id)
-                    if vulnerability_ids:
-                        finding.unsaved_vulnerability_ids = vulnerability_ids
-
-                    findings.append(finding)
+        findings = None
+        # this parser can handle two distinct formats see sample scan files
+        if "dependencies" in data:
+            # new format of report
+            findings = get_file_findings(data, test)
+        else:
+            # legacy format of report
+            findings = get_legacy_findings(data, test)
 
         return findings
+
+
+def get_file_findings(data, test):
+    """Return the findings in the vluns array inside the dependencies key."""
+    findings = list()
+    for dependency in data["dependencies"]:
+        item_findings = get_item_findings(dependency, test)
+        if item_findings is not None:
+            findings.extend(item_findings)
+    return findings
+
+
+def get_legacy_findings(data, test):
+    """Return the findings gathered from the vulns element."""
+    findings = list()
+    for item in data:
+        item_findings = get_item_findings(item, test)
+        if item_findings is not None:
+            findings.extend(item_findings)
+    return findings
+
+
+def get_item_findings(item, test):
+    """Return list of Findings."""
+    findings = list()
+    vulnerabilities = item.get("vulns", [])
+    if vulnerabilities:
+        component_name = item["name"]
+        component_version = item.get("version")
+        for vulnerability in vulnerabilities:
+            vuln_id = vulnerability.get("id")
+            vuln_fix_versions = vulnerability.get("fix_versions")
+            vuln_description = vulnerability.get("description")
+
+            title = (
+                f"{vuln_id} in {component_name}:{component_version}"
+            )
+
+            description = ""
+            description += vuln_description
+
+            mitigation = None
+            if vuln_fix_versions:
+                mitigation = "Upgrade to version:"
+                if len(vuln_fix_versions) == 1:
+                    mitigation += f" {vuln_fix_versions[0]}"
+                else:
+                    for fix_version in vuln_fix_versions:
+                        mitigation += f"\n- {fix_version}"
+
+            finding = Finding(
+                test=test,
+                title=title,
+                cwe=1395,
+                severity="Medium",
+                description=description,
+                mitigation=mitigation,
+                component_name=component_name,
+                component_version=component_version,
+                vuln_id_from_tool=vuln_id,
+                static_finding=True,
+                dynamic_finding=False,
+            )
+            vulnerability_ids = list()
+            if vuln_id:
+                vulnerability_ids.append(vuln_id)
+            if vulnerability_ids:
+                finding.unsaved_vulnerability_ids = vulnerability_ids
+
+            findings.append(finding)
+
+    return findings

unittests/scans/pip_audit/empty_new.json

@@ -0,0 +1,3 @@
+{
+  "dependencies":[]
+}

unittests/scans/pip_audit/many_vulns_new.json

@@ -0,0 +1,91 @@
+{
+      "dependencies":[
+        {
+            "name": "adal",
+            "version": "1.2.2",
+            "vulns": []
+        },
+        {
+            "name": "aiohttp",
+            "version": "3.6.2",
+            "vulns": [
+                {
+                    "id": "PYSEC-2021-76",
+                    "fix_versions": [
+                        "3.7.4"
+                    ],
+                    "description": "aiohttp is an asynchronous HTTP client/server framework for asyncio and Python. In aiohttp before version 3.7.4 there is an open redirect vulnerability. A maliciously crafted link to an aiohttp-based web-server could redirect the browser to a different website. It is caused by a bug in the `aiohttp.web_middlewares.normalize_path_middleware` middleware. This security problem has been fixed in 3.7.4. Upgrade your dependency using pip as follows \"pip install aiohttp >= 3.7.4\". If upgrading is not an option for you, a workaround can be to avoid using `aiohttp.web_middlewares.normalize_path_middleware` in your applications."
+                }
+            ]
+        },
+        {
+            "name": "alabaster",
+            "version": "0.7.12",
+            "vulns": []
+        },
+        {
+            "name": "azure-devops",
+            "skip_reason": "Dependency not found on PyPI and could not be audited: azure-devops (0.17.0)"
+        },
+        {
+            "name": "django",
+            "version": "3.2.9",
+            "vulns": [
+                {
+                    "id": "PYSEC-2021-439",
+                    "fix_versions": [
+                        "2.2.25",
+                        "3.1.14",
+                        "3.2.10"
+                    ],
+                    "description": "In Django 2.2 before 2.2.25, 3.1 before 3.1.14, and 3.2 before 3.2.10, HTTP requests for URLs with trailing newlines could bypass upstream access control based on URL paths."
+                }
+            ]
+        },
+        {
+            "name": "lxml",
+            "version": "4.6.4",
+            "vulns": [
+                {
+                    "id": "PYSEC-2021-852",
+                    "fix_versions": [],
+                    "description": "lxml is a library for processing XML and HTML in the Python language. Prior to version 4.6.5, the HTML Cleaner in lxml.html lets certain crafted script content pass through, as well as script content in SVG files embedded using data URIs. Users that employ the HTML cleaner in a security relevant context should upgrade to lxml 4.6.5 to receive a patch. There are no known workarounds available."
+                }
+            ]
+        },
+        {
+            "name": "twisted",
+            "version": "18.9.0",
+            "vulns": [
+                {
+                    "id": "PYSEC-2019-128",
+                    "fix_versions": [
+                        "19.2.1"
+                    ],
+                    "description": "In Twisted before 19.2.1, twisted.web did not validate or sanitize URIs or HTTP methods, allowing an attacker to inject invalid characters such as CRLF."
+                },
+                {
+                    "id": "PYSEC-2020-260",
+                    "fix_versions": [
+                        "20.3.0rc1"
+                    ],
+                    "description": "In Twisted Web through 19.10.0, there was an HTTP request splitting vulnerability. When presented with a content-length and a chunked encoding header, the content-length took precedence and the remainder of the request body was interpreted as a pipelined request."
+                },
+                {
+                    "id": "PYSEC-2019-129",
+                    "fix_versions": [
+                        "19.7.0rc1"
+                    ],
+                    "description": "In words.protocols.jabber.xmlstream in Twisted through 19.2.1, XMPP support did not verify certificates when used with TLS, allowing an attacker to MITM connections."
+                },
+                {
+                    "id": "PYSEC-2020-259",
+                    "fix_versions": [
+                        "20.3.0rc1"
+                    ],
+                    "description": "In Twisted Web through 19.10.0, there was an HTTP request splitting vulnerability. When presented with two content-length headers, it ignored the first header. When the second content-length value was set to zero, the request body was interpreted as a pipelined request."
+                }
+            ]
+        }
+    ]
+}

unittests/scans/pip_audit/zero_vulns_new.json

@@ -0,0 +1,18 @@
+{
+      "dependencies":[
+        {
+            "name": "adal",
+            "version": "1.2.2",
+            "vulns": []
+        },
+        {
+            "name": "alabaster",
+            "version": "0.7.12",
+            "vulns": []
+        },
+        {
+            "name": "azure-devops",
+            "skip_reason": "Dependency not found on PyPI and could not be audited: azure-devops (0.17.0)"
+        }
+    ]
+}