update parsers that didn't parse cvss correctly yet

Author: valentijnscholten

docs/content/en/open_source/contributing/how-to-write-a-parser.md

@@ -169,14 +169,22 @@ Good example:
 ### Do not parse CVSS by hand (vector, score or severity)
 
 Data can have `CVSS` vectors or scores. Don't write your own CVSS score algorithm.
-For parser, we rely on module `cvss`.
+For parser, we rely on module `cvss`. But we also have a helper method to validate the vector and extract the base score and severity from it.
 
-It's easy to use and will make the parser aligned with the rest of the code.
+```python
+cvss_data = parse_cvss_data("CVSS:3.0/S:C/C:H/I:H/A:N/AV:P/AC:H/PR:H/UI:R/E:H/RL:O/RC:R/CR:H/IR:X/AR:X/MAC:H/MPR:X/MUI:X/MC:L/MA:X")
+if cvss_data:
+    finding.cvss3 = cvss_data.get("vector")
+    finding.cvssv3_score = cvss_data.get("score")
+    finding.severity = cvss_data.get("severity")  # if your tool does generate severity
+```
+
+If you need more manual processing, you can parse the `CVSS` vector directly.
 
 Example of use:
 
 ```python
-from cvss.cvss3 import CVSS3
+from dojo.utils import cvss.cvss3 import CVSS3
 import cvss.parser
 vectors = cvss.parser.parse_cvss_from_text("CVSS:3.0/S:C/C:H/I:H/A:N/AV:P/AC:H/PR:H/UI:R/E:H/RL:O/RC:R/CR:H/IR:X/AR:X/MAC:H/MPR:X/MUI:X/MC:L/MA:X")
 if len(vectors) > 0 and type(vectors[0]) is CVSS3:
@@ -186,17 +194,8 @@ if len(vectors) > 0 and type(vectors[0]) is CVSS3:
     severity = vectors[0].severities()[0]
     vectors[0].compute_base_score()
     cvssv3_score = vectors[0].scores()[0]
-    print(severity)
-    print(cvssv3_score)
-```
-
-Good example:
-
-```python
-vectors = cvss.parser.parse_cvss_from_text(item['cvss_vect'])
-if len(vectors) > 0 and type(vectors[0]) is CVSS3:
-    finding.cvss = vectors[0].clean_vector()
-    finding.severity = vectors[0].severities()[0]  # if your tool does generate severity
+    finding.severity = severity
+    finding.cvssv3_score = cvssv3_score
 ```
 
 Bad example (DIY):

dojo/models.py

@@ -10,11 +10,11 @@
 from pathlib import Path
 from uuid import uuid4
 
+import cvss.parser
 import dateutil
 import hyperlink
 import tagulous.admin
 from auditlog.registry import auditlog
-from cvss import CVSS3
 from dateutil.relativedelta import relativedelta
 from django import forms
 from django.conf import settings
@@ -2333,8 +2333,6 @@ class Finding(models.Model):
                               verbose_name=_("EPSS percentile"),
                               help_text=_("EPSS percentile for the CVE. Describes how many CVEs are scored at or below this one."),
                               validators=[MinValueValidator(0.0), MaxValueValidator(1.0)])
-    # cvssv3_regex = RegexValidator(regex=r"^AV:[NALP]|AC:[LH]|PR:[UNLH]|UI:[NR]|S:[UC]|[CIA]:[NLH]", message="CVSS must be entered in format: 'AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:H/A:H'")
-    # cvssv3 = models.TextField(validators=[cvssv3_regex],
     cvssv3 = models.TextField(validators=[cvss3_validator],
                               max_length=117,
                               null=True,
@@ -2702,11 +2700,12 @@ def save(self, dedupe_option=True, rules_option=True, product_grading_option=Tru
         # Synchronize cvssv3 score using cvssv3 vector
         if self.cvssv3:
             try:
-                cvss_object = CVSS3(self.cvssv3)
+                cvss_vector = cvss.parser.parse_cvss_from_text(self.cvssv3)
                 # use the environmental score, which is the most refined score
-                self.cvssv3_score = cvss_object.scores()[2]
+                self.cvssv3_score = cvss_vector.scores()[2]
             except Exception as ex:
-                logger.error("Can't compute cvssv3 score for finding id %i. Invalid cvssv3 vector found: '%s'. Exception: %s", self.id, self.cvssv3, ex)
+                logger.warning("Can't compute cvssv3 score for finding id %i. Invalid cvssv3 vector found: '%s'. Exception: %s.", self.id, self.cvssv3, ex)
+                # should we set self.cvssv3 to None here to avoid storing invalid vectors? it would also remove invalid vectors on existing findings...
 
         self.set_hash_code(dedupe_option)
 
@@ -3519,8 +3518,6 @@ class Finding_Template(models.Model):
                            blank=False,
                            verbose_name="Vulnerability Id",
                            help_text="An id of a vulnerability in a security advisory associated with this finding. Can be a Common Vulnerabilities and Exposures (CVE) or from other sources.")
-    # cvssv3_regex = RegexValidator(regex=r"^AV:[NALP]|AC:[LH]|PR:[UNLH]|UI:[NR]|S:[UC]|[CIA]:[NLH]", message="CVSS must be entered in format: 'AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:H/A:H'")
-    # cvssv3 = models.TextField(validators=[cvssv3_regex], max_length=117, null=True)
     cvssv3 = models.TextField(help_text=_("Common Vulnerability Scoring System version 3 (CVSSv3) score associated with this finding."), validators=[cvss3_validator], max_length=117, null=True, verbose_name=_("CVSS v3 vector"))
 
     severity = models.CharField(max_length=200, null=True, blank=True)

dojo/tools/aqua/parser.py

@@ -1,6 +1,7 @@
 import json
 
 from dojo.models import Finding
+from dojo.utils import parse_cvss_data
 
 
 class AquaParser:
@@ -170,7 +171,7 @@ def get_item(self, resource, vuln, test):
             severity_justification += "\nAqua severity classification: {}".format(vuln.get("aqua_severity_classification"))
             severity_justification += "\nAqua scoring system: {}".format(vuln.get("aqua_scoring_system"))
             if "nvd_score_v3" in vuln:
-                cvssv3 = vuln.get("nvd_vectors_v3")  # TODO: VECTOR
+                cvssv3 = vuln.get("nvd_vectors_v3")
         if "aqua_score" in vuln:
             if score is None:
                 score = vuln.get("aqua_score")
@@ -193,14 +194,15 @@ def get_item(self, resource, vuln, test):
                 )
             severity_justification += "\nNVD v3 vectors: {}".format(vuln.get("nvd_vectors_v3"))
             # Add the CVSS3 to Finding
-            cvssv3 = vuln.get("nvd_vectors_v3")  # TODO: VECTOR
+            cvssv3 = vuln.get("nvd_vectors_v3")
         if "nvd_score" in vuln:
             if score is None:
                 score = vuln.get("nvd_score")
                 used_for_classification = (
                     f"NVD score v2 ({score}) used for classification.\n"
                 )
             severity_justification += "\nNVD v2 vectors: {}".format(vuln.get("nvd_vectors"))
+
         severity_justification += f"\n{used_for_classification}"
         severity = self.severity_of(score)
         finding = Finding(
@@ -214,14 +216,19 @@ def get_item(self, resource, vuln, test):
             severity=severity,
             severity_justification=severity_justification,
             cwe=0,
-            cvssv3=cvssv3,
             description=description.strip(),
             mitigation=fix_version,
             references=url,
             component_name=resource.get("name"),
             component_version=resource.get("version"),
             impact=severity,
         )
+
+        cvss_data = parse_cvss_data(cvssv3)
+        if cvss_data:
+            finding.cvss3 = cvss_data.get("vector")
+            finding.cvssv3_score = cvss_data.get("score")
+
         if vulnerability_id != "No CVE":
             finding.unsaved_vulnerability_ids = [vulnerability_id]
         if vuln.get("epss_score"):

dojo/tools/blackduck_binary_analysis/parser.py

@@ -43,7 +43,7 @@ def ingest_findings(self, sorted_findings, test):
             cwe = 1357
             title = self.format_title(i)
             description = self.format_description(i)
-            cvss_v3 = True  # TODO: VECTOR
+            cvss_v3 = True
             if str(i.cvss_vector_v3) != "":
                 cvss_vectors = "{}{}".format(
                     "CVSS:3.1/",

dojo/tools/cyberwatch_galeax/parser.py

@@ -197,7 +197,7 @@ def build_findings_for_cve(self, cve_code, c_data, test):
         description = c_data["description"]
         impact = c_data["impact"]
         references = c_data["references"]
-        cvssv3 = c_data["cvssv3"]  # TODO: VECTOR
+        cvssv3 = c_data["cvssv3"]
         cvssv3_score = c_data["cvssv3_score"]
         products = c_data["products"]
 
@@ -515,7 +515,7 @@ def parse_cvss(self, cvss_v3_vector, json_data):
         if cvss_v3_vector:
             vectors = cvss.parser.parse_cvss_from_text(cvss_v3_vector)
             if vectors and isinstance(vectors[0], CVSS3):
-                cvssv3 = vectors[0].clean_vector()  # TODO: VECTOR
+                cvssv3 = vectors[0].clean_vector()
                 cvssv3_score = vectors[0].scores()[0]
                 severity = vectors[0].severities()[0]
                 return cvssv3, cvssv3_score, severity

dojo/tools/jfrog_xray_unified/parser.py

@@ -2,6 +2,7 @@
 from datetime import datetime
 
 from dojo.models import Finding
+from dojo.utils import parse_cvss_data
 
 
 class JFrogXrayUnifiedParser:
@@ -75,7 +76,7 @@ def get_item(vulnerability, test):
             vulnerability_id = worstCve["cve"]
         if "cvss_v3_vector" in worstCve:
             cvss_v3 = worstCve["cvss_v3_vector"]
-            cvssv3 = cvss_v3  # TODO: VECTOR
+            cvssv3 = cvss_v3
         if "cvss_v2_vector" in worstCve:
             cvss_v2 = worstCve["cvss_v2_vector"]
 
@@ -134,12 +135,16 @@ def get_item(vulnerability, test):
         dynamic_finding=False,
         references=references,
         impact=severity,
-        cvssv3=cvssv3,
         date=scan_time,
         unique_id_from_tool=vulnerability["issue_id"],
         tags=tags,
     )
 
+    cvss_data = parse_cvss_data(cvssv3)
+    if cvss_data:
+        finding.cvss3 = cvss_data.get("vector")
+        finding.cvssv3_score = cvss_data.get("score")
+
     if vulnerability_id:
         finding.unsaved_vulnerability_ids = [vulnerability_id]
 

dojo/tools/npm_audit_7_plus/parser.py

@@ -165,8 +165,12 @@ def get_item(item_node, tree, test):
         cwe = int(cwe.split("-")[1])
         dojo_finding.cwe = cwe
 
-    if (cvssv3 is not None) and (len(cvssv3) > 0):  # TODO: VECTOR
-        dojo_finding.cvssv3 = cvssv3
+    if (cvssv3 is not None) and (len(cvssv3) > 0):
+        from dojo.utils import parse_cvss_data
+        cvss_data = parse_cvss_data(cvssv3)
+        if cvss_data:
+            dojo_finding.cvss3 = cvss_data.get("vector")
+            dojo_finding.cvssv3_score = cvss_data.get("score")
 
     return dojo_finding
 

dojo/tools/qualys/csv_parser.py

@@ -8,6 +8,7 @@
 from django.conf import settings
 
 from dojo.models import Endpoint, Finding
+from dojo.utils import parse_cvss_data
 
 _logger = logging.getLogger(__name__)
 
@@ -197,7 +198,7 @@ def build_findings_from_dict(report_findings: [dict]) -> [Finding]:
         # Clean up the CVE data appropriately
         cve_list = _clean_cve_data(cve_data)
 
-        if "CVSS3 Base" in report_finding:  # TODO: VECTOR
+        if "CVSS3 Base" in report_finding:
             cvssv3 = _extract_cvss_vectors(
                         report_finding["CVSS3 Base"], report_finding["CVSS3 Temporal"],
                     )
@@ -227,8 +228,13 @@ def build_findings_from_dict(report_findings: [dict]) -> [Finding]:
                 impact=report_finding["Impact"],
                 date=date,
                 vuln_id_from_tool=report_finding["QID"],
-                cvssv3=cvssv3,
             )
+            # Make sure vector is valid
+            cvss_data = parse_cvss_data(cvssv3)
+            if cvss_data:
+                finding.cvss3 = cvss_data.get("vector")
+                finding.cvssv3_score = cvss_data.get("score")
+
             # Qualys reports regression findings as active, but with a Date Last
             # Fixed.
             if report_finding["Date Last Fixed"]:

dojo/tools/qualys/parser.py

@@ -8,6 +8,7 @@
 
 from dojo.models import Endpoint, Finding
 from dojo.tools.qualys import csv_parser
+from dojo.utils import parse_cvss_data
 
 logger = logging.getLogger(__name__)
 
@@ -352,7 +353,11 @@ def parse_finding(host, tree):
         finding.is_mitigated = temp["mitigated"]
         finding.active = temp["active"]
         if temp.get("CVSS_vector") is not None:
-            finding.cvssv3 = temp.get("CVSS_vector")  # TODO: VECTOR
+            cvss_data = parse_cvss_data(temp.get("CVSS_vector"))
+            if cvss_data:
+                finding.cvss3 = cvss_data.get("vector")
+                finding.cvss3_score = cvss_data.get("base_score")
+
         if temp.get("CVSS_value") is not None:
             finding.cvssv3_score = temp.get("CVSS_value")
         finding.verified = True

dojo/tools/sonatype/parser.py

@@ -2,6 +2,7 @@
 
 from dojo.models import Finding
 from dojo.tools.sonatype.identifier import ComponentIdentifier
+from dojo.utils import parse_cvss_data
 
 
 class SonatypeParser:
@@ -63,7 +64,10 @@ def get_finding(security_issue, component, test):
         finding.cwe = security_issue["cwe"]
 
     if "cvssVector" in security_issue:
-        finding.cvssv3 = security_issue["cvssVector"]  # TODO: VECTOR
+        cvss_data = parse_cvss_data(security_issue["cvssVector"])
+        if cvss_data:
+            finding.cvss3 = cvss_data.get("vector")
+            finding.cvssv3_score = cvss_data.get("score")
 
     if "pathnames" in component:
         finding.file_path = " ".join(component["pathnames"])[:1000]