Merge pull request #747 from ziadhany/calculate_cvss_vector

add support for calculating CVSS score from the CVSS vector

Author: pombredanne

pyproject.toml

@@ -74,11 +74,19 @@ addopts = [
 line-length = 100
 include = '\.pyi?$'
 skip_gitignore = true
-extend-exclude = "migrations|data|venv"
+# 'extend-exclude' excludes files or directories in addition to the defaults
+extend-exclude = '''
+(
+  ^/venv/.*
+  | ^/vulnerabilities/migrations/.*
+  | ^/vulnerabilities/tests/test_data/.*
+)
+'''
+
 
 [tool.isort]
 profile = "black"
 line_length = 100
 force_single_line = true
 skip_gitignore = true
-skip_glob = "*/migrations/*"
+skip_glob = "vulnerabilities/migrations/*"

setup.cfg

@@ -83,6 +83,7 @@ install_requires =
     defusedxml>=0.7.1
     Markdown>=3.3.0
     dateparser>=1.1.1
+    cvss>=2.4
 
     # networking
     GitPython>=3.1.17

vulnerabilities/api.py

@@ -29,7 +29,7 @@
 class VulnerabilitySeveritySerializer(serializers.ModelSerializer):
     class Meta:
         model = VulnerabilitySeverity
-        fields = ["value", "scoring_system"]
+        fields = ["value", "scoring_system", "scoring_elements"]
 
 
 class VulnerabilityReferenceSerializer(serializers.ModelSerializer):

vulnerabilities/importer.py

@@ -46,13 +46,16 @@
 
 @dataclasses.dataclass(order=True)
 class VulnerabilitySeverity:
+    # FIXME: this should be named scoring_system, like in the model
     system: ScoringSystem
     value: str
+    scoring_elements: str = ""
 
     def to_dict(self):
         return {
             "system": self.system.identifier,
             "value": self.value,
+            "scoring_elements": self.scoring_elements,
         }
 
     @classmethod
@@ -61,7 +64,11 @@ def from_dict(cls, severity: dict):
         Return a VulnerabilitySeverity object from a ``severity`` mapping of
         VulnerabilitySeverity data.
         """
-        return cls(system=SCORING_SYSTEMS[severity["system"]], value=severity["value"])
+        return cls(
+            system=SCORING_SYSTEMS[severity["system"]],
+            value=severity["value"],
+            scoring_elements=severity.get("scoring_elements", ""),
+        )
 
 
 @dataclasses.dataclass(order=True)
@@ -426,15 +433,13 @@ def get_data_from_xml_doc(
             # connected/linked to an OvalDefinition
             vuln_id = definition_data["vuln_id"]
             description = definition_data["description"]
-            severities = (
-                [
-                    VulnerabilitySeverity(
-                        system=severity_systems.GENERIC, value=definition_data.get("severity")
-                    )
-                ]
-                if definition_data.get("severity")
-                else []
-            )
+
+            severities = []
+            severity = definition_data.get("severity")
+            if severity:
+                severities.append(
+                    VulnerabilitySeverity(system=severity_systems.GENERIC, value=severity)
+                )
             references = [
                 Reference(url=url, severities=severities)
                 for url in definition_data["reference_urls"]

vulnerabilities/importers/nvd.py

@@ -167,12 +167,7 @@ def severities(self):
             vs = VulnerabilitySeverity(
                 system=severity_systems.CVSSV3,
                 value=str(cvss_v3.get("baseScore") or ""),
-            )
-            severities.append(vs)
-
-            vs = VulnerabilitySeverity(
-                system=severity_systems.CVSSV3_VECTOR,
-                value=str(cvss_v3.get("vectorString") or ""),
+                scoring_elements=str(cvss_v3.get("vectorString") or ""),
             )
             severities.append(vs)
 
@@ -182,12 +177,7 @@ def severities(self):
             vs = VulnerabilitySeverity(
                 system=severity_systems.CVSSV2,
                 value=str(cvss_v2.get("baseScore") or ""),
-            )
-            severities.append(vs)
-
-            vs = VulnerabilitySeverity(
-                system=severity_systems.CVSSV2_VECTOR,
-                value=str(cvss_v2.get("vectorString") or ""),
+                scoring_elements=str(cvss_v2.get("vectorString") or ""),
             )
             severities.append(vs)
 

vulnerabilities/importers/osv.py

@@ -115,10 +115,10 @@ def get_severities(raw_data) -> Iterable[VulnerabilitySeverity]:
     """
     for severity in raw_data.get("severity") or []:
         if severity.get("type") == "CVSS_V3":
-            yield VulnerabilitySeverity(
-                system=SCORING_SYSTEMS["cvssv3.1_vector"],
-                value=severity["score"],
-            )
+            vector = severity["score"]
+            system = SCORING_SYSTEMS["cvssv3.1"]
+            score = system.compute(vector)
+            yield VulnerabilitySeverity(system=system, value=score, scoring_elements=vector)
         else:
             logger.error(f"Unsupported severity type: {severity!r} for OSV id: {raw_data['id']!r}")
 

vulnerabilities/importers/postgresql.py

@@ -95,16 +95,12 @@ def to_advisories(data):
                     parsed_link = urlparse.urlparse(vector_link_tag["href"])
                     cvss3_vector = urlparse.parse_qs(parsed_link.query)["vector"]
                     cvss3_base_score = vector_link_tag.text
-                    severities.extend(
-                        [
-                            VulnerabilitySeverity(
-                                system=severity_systems.CVSSV3, value=cvss3_base_score
-                            ),
-                            VulnerabilitySeverity(
-                                system=severity_systems.CVSSV3_VECTOR, value=cvss3_vector
-                            ),
-                        ]
+                    severity = VulnerabilitySeverity(
+                        system=severity_systems.CVSSV3,
+                        value=cvss3_base_score,
+                        scoring_elements=cvss3_vector,
                     )
+                    severities.append(severity)
             references.append(Reference(url=link, severities=severities))
 
         advisories.append(

vulnerabilities/importers/redhat.py

@@ -28,22 +28,23 @@
 
 logger = logging.getLogger(__name__)
 
+# FIXME: we should use a centralized retry
 requests_session = requests_with_5xx_retry(max_retries=5, backoff_factor=1)
 
 
-def fetch_list_of_cves() -> Iterable[List[Dict]]:
+def fetch_cves() -> Iterable[List[Dict]]:
     page_no = 1
     cve_data = None
     while True:
         current_url = f"https://access.redhat.com/hydra/rest/securitydata/cve.json?per_page=1000&page={page_no}"  # nopep8
         try:
             response = requests_session.get(current_url)
             if response.status_code != requests.codes.ok:
-                logger.error(f"Failed to fetch results from {current_url}")
+                logger.error(f"Failed to fetch RedHat CVE results from {current_url}")
                 break
             cve_data = response.json()
         except Exception as e:
-            logger.error(f"Failed to fetch results from {current_url} {e}")
+            logger.error(f"Failed to fetch RedHat CVE results from {current_url} {e}")
             break
         if not cve_data:
             break
@@ -65,8 +66,8 @@ class RedhatImporter(Importer):
     license_url = "https://access.redhat.com/documentation/en-us/red_hat_security_data_api/1.0/html/red_hat_security_data_api/legal-notice"
 
     def advisory_data(self) -> Iterable[AdvisoryData]:
-        for list_of_redhat_cves in fetch_list_of_cves():
-            for redhat_cve in list_of_redhat_cves:
+        for redhat_cves in fetch_cves():
+            for redhat_cve in redhat_cves:
                 yield to_advisory(redhat_cve)
 
 
@@ -154,20 +155,13 @@ def to_advisory(advisory_data):
 
     redhat_scores = []
     cvssv3_score = advisory_data.get("cvss3_score")
+    cvssv3_vector = advisory_data.get("cvss3_scoring_vector", "")
     if cvssv3_score:
         redhat_scores.append(
             VulnerabilitySeverity(
                 system=severity_systems.CVSSV3,
                 value=cvssv3_score,
-            )
-        )
-
-    cvssv3_vector = advisory_data.get("cvss3_scoring_vector")
-    if cvssv3_vector:
-        redhat_scores.append(
-            VulnerabilitySeverity(
-                system=severity_systems.CVSSV3_VECTOR,
-                value=cvssv3_vector,
+                scoring_elements=cvssv3_vector,
             )
         )
 

vulnerabilities/importers/suse_scores.py

@@ -26,37 +26,26 @@ def updated_advisories(self):
 
     @staticmethod
     def to_advisory(score_data):
+        systems_by_version = {
+            "2.0": severity_systems.CVSSV2,
+            "3": severity_systems.CVSSV3,
+            "3.1": severity_systems.CVSSV31,
+        }
         advisories = []
+
         for cve_id in score_data:
             severities = []
             for cvss_score in score_data[cve_id]["cvss"]:
-                score = None
-                vector = None
-                if cvss_score["version"] == "2.0":
-                    score = VulnerabilitySeverity(
-                        system=severity_systems.CVSSV2, value=str(cvss_score["score"])
-                    )
-                    vector = VulnerabilitySeverity(
-                        system=severity_systems.CVSSV2_VECTOR, value=str(cvss_score["vector"])
-                    )
-
-                elif cvss_score["version"] == "3":
-                    score = VulnerabilitySeverity(
-                        system=severity_systems.CVSSV3, value=str(cvss_score["score"])
-                    )
-                    vector = VulnerabilitySeverity(
-                        system=severity_systems.CVSSV3_VECTOR, value=str(cvss_score["vector"])
-                    )
-
-                elif cvss_score["version"] == "3.1":
-                    score = VulnerabilitySeverity(
-                        system=severity_systems.CVSSV31, value=str(cvss_score["score"])
-                    )
-                    vector = VulnerabilitySeverity(
-                        system=severity_systems.CVSSV31_VECTOR, value=str(cvss_score["vector"])
-                    )
-
-                severities.extend([score, vector])
+                cvss_version = cvss_score["version"]
+                scoring_system = systems_by_version[cvss_version]
+                base_score = str(cvss_score["score"])
+                vector = str(cvss_score.get("vector", ""))
+                score = VulnerabilitySeverity(
+                    system=scoring_system,
+                    value=base_score,
+                    scoring_elements=vector,
+                )
+                severities.append(score)
 
             advisories.append(
                 AdvisoryData(

vulnerabilities/improve_runner.py

@@ -102,10 +102,16 @@ def process_inferences(inferences: List[Inference], advisory: Advisory, improver
                 _vs, updated = VulnerabilitySeverity.objects.update_or_create(
                     scoring_system=severity.system.identifier,
                     reference=reference,
-                    defaults={"value": str(severity.value)},
+                    defaults={
+                        "value": str(severity.value),
+                        "scoring_elements": str(severity.scoring_elements),
+                    },
                 )
                 if updated:
-                    logger.info(f"Severity updated for reference {ref!r} to {severity.value!r}")
+                    logger.info(
+                        f"Severity updated for reference {ref!r} to value: {severity.value!r} "
+                        f"and scoring_elements: {severity.scoring_elements!r}"
+                    )
 
         for affected_purl in inference.affected_purls or []:
             vulnerable_package = Package.objects.get_or_create_from_purl(purl=affected_purl)