Store severity scores

.gitignore

@@ -122,3 +122,6 @@ Pipfile
 
 # pytest
 .pytest_cache
+
+# VSCode
+.vscode

vulnerabilities/admin.py

@@ -29,6 +29,7 @@
  Package,
  Vulnerability,
  VulnerabilityReference,
+ VulnerabilitySeverity,
 )
 
 
@@ -57,3 +58,8 @@ class PackageRelatedVulnerabilityAdmin(admin.ModelAdmin):
 @admin.register(Importer)
 class ImporterAdmin(admin.ModelAdmin):
  pass
+
+
+@admin.register(VulnerabilitySeverity)
+class VulnerabilitySeverityAdmin(admin.ModelAdmin):
+ pass

vulnerabilities/fixtures/debian.json

@@ -4,26 +4,26 @@
  "pk": 1,
  "fields": {
  "cve_id": "CVE-2014-8242",
- "summary": "",
-"cvss": null
+ "summary": ""
+
  }
 },
 {
  "model": "vulnerabilities.vulnerability",
  "pk": 2,
  "fields": {
  "cve_id": "CVE-2009-1382",
- "summary": "",
-"cvss": null
+ "summary": ""
+
  }
 },
 {
  "model": "vulnerabilities.vulnerability",
  "pk": 3,
  "fields": {
  "cve_id": "CVE-2009-2459",
- "summary": "",
-"cvss": null
+ "summary": ""
+
  }
 },
 {

vulnerabilities/import_runner.py

@@ -41,27 +41,7 @@
 
 logger = logging.getLogger(__name__)
 
-# These _inserter classes are used to instantiate model objects.
-# Frozen dataclass store args required to store instantiate
-# model objects, this way model objects can be hashed indirectly which
-# is required in this implementation.
-
-
-@dataclasses.dataclass(frozen=True)
-class VulnerabilityReferenceInserter:
- vulnerability: models.Vulnerability
- reference_id: Optional[str] = ""
- url: Optional[str] = ""
-
- def __post_init__(self):
- if not any([self.reference_id, self.url]):
- raise TypeError("VulnerabilityReferenceInserter expects either reference_id or url")
-
- def to_model_object(self):
- return models.VulnerabilityReference(**dataclasses.asdict(self))
-
-
-# These _inserter classes are used to instantiate model objects.
+# This *Inserter class is used to instantiate model objects.
 # Frozen dataclass store args required to store instantiate
 # model objects, this way model objects can be hashed indirectly which
 # is required in this implementation.
@@ -134,7 +114,6 @@ def get_vuln_pkg_refs(vulnerability, package):
 
 
 def process_advisories(data_source: DataSource) -> None:
- bulk_create_vuln_refs = set()
  bulk_create_vuln_pkg_refs = set()
  # Treat updated_advisories and added_advisories as same. Eventually
  # we want to refactor all data sources to provide advisories via a
@@ -145,18 +124,17 @@ def process_advisories(data_source: DataSource) -> None:
  try:
  vuln, vuln_created = _get_or_create_vulnerability(advisory)
  for vuln_ref in advisory.vuln_references:
- ref = VulnerabilityReferenceInserter(
- vulnerability=vuln,
- url=vuln_ref.url,
- reference_id=vuln_ref.reference_id,
+ ref, _ = models.VulnerabilityReference.objects.get_or_create(
+ vulnerability=vuln, reference_id=vuln_ref.reference_id, url=vuln_ref.url
  )
 
- if vuln_created or not vuln_ref_exists(
- vuln, vuln_ref.url, vuln_ref.reference_id
- ):
- # A vulnerability reference can't exist if the
- # vulnerability is just created so insert it
- bulk_create_vuln_refs.add(ref)
+ for score in vuln_ref.severities:
+ models.VulnerabilitySeverity.objects.update_or_create(
+ vulnerability=vuln,
+ scoring_system=score.system.identifier,
+ reference=ref,
+ defaults={"value": str(score.value)},
+ )
 
  for purl in chain(advisory.impacted_package_urls, advisory.resolved_package_urls):
  pkg, pkg_created = _get_or_create_package(purl)
@@ -184,14 +162,9 @@ def process_advisories(data_source: DataSource) -> None:
  except Exception:
  # TODO: store error but continue
  logger.error(
- f"Failed to process advisory: {advisory!r}:\n"
- + traceback.format_exc()
+ f"Failed to process advisory: {advisory!r}:\n" + traceback.format_exc()
  )
 
- models.VulnerabilityReference.objects.bulk_create(
- [i.to_model_object() for i in bulk_create_vuln_refs]
- )
-
  # find_conflicting_relations handles in-memory conflicts
  conflicts = find_conflicting_relations(bulk_create_vuln_pkg_refs)
 
@@ -267,8 +240,8 @@ def _get_or_create_vulnerability(
 
  except Exception:
  logger.error(
- f"Failed to _get_or_create_vulnerability: {query_kwargs!r}:\n"
- + traceback.format_exc())
+ f"Failed to _get_or_create_vulnerability: {query_kwargs!r}:\n" + traceback.format_exc()
+ )
  raise
 
 

vulnerabilities/importers/nvd.py

@@ -32,7 +32,9 @@
 from vulnerabilities.data_source import DataSource
 from vulnerabilities.data_source import DataSourceConfiguration
 from vulnerabilities.data_source import Reference
+from vulnerabilities.data_source import VulnerabilitySeverity
 from vulnerabilities.helpers import create_etag
+from vulnerabilities.severity_systems import scoring_systems
 
 
 @dataclasses.dataclass
@@ -77,6 +79,14 @@ def to_advisories(self, nvd_data):
  cve_id = cve_item["cve"]["CVE_data_meta"]["ID"]
  ref_urls = self.extract_reference_urls(cve_item)
  references = [Reference(url=url) for url in ref_urls]
+ severity_scores = self.extract_severity_scores(cve_item)
+ references.append(
+ Reference(
+ url=f"https://nvd.nist.gov/vuln/detail/{cve_id}",
+ reference_id=cve_id,
+ severities=severity_scores,
+ )
+ )
  summary = self.extract_summary(cve_item)
  yield Advisory(
  cve_id=cve_id, summary=summary, vuln_references=references, impacted_package_urls=[]
@@ -90,6 +100,40 @@ def extract_summary(cve_item):
  summaries = [desc["value"] for desc in cve_item["cve"]["description"]["description_data"]]
  return max(summaries, key=len)
 
+ @staticmethod
+ def extract_severity_scores(cve_item):
+ severity_scores = []
+
+ if cve_item["impact"].get("baseMetricV3"):
+ severity_scores.append(
+ VulnerabilitySeverity(
+ system=scoring_systems["cvssv3"],
+ value=str(cve_item["impact"]["baseMetricV3"]["cvssV3"]["baseScore"]),
+ )
+ )
+ severity_scores.append(
+ VulnerabilitySeverity(
+ system=scoring_systems["cvssv3_vector"],
+ value=str(cve_item["impact"]["baseMetricV3"]["cvssV3"]["vectorString"]),
+ )
+ )
+
+ if cve_item["impact"].get("baseMetricV2"):
+ severity_scores.append(
+ VulnerabilitySeverity(
+ system=scoring_systems["cvssv2"],
+ value=str(cve_item["impact"]["baseMetricV2"]["cvssV2"]["baseScore"]),
+ )
+ )
+ severity_scores.append(
+ VulnerabilitySeverity(
+ system=scoring_systems["cvssv2_vector"],
+ value=str(cve_item["impact"]["baseMetricV2"]["cvssV2"]["vectorString"]),
+ )
+ )
+
+ return severity_scores
+
  def extract_reference_urls(self, cve_item):
  urls = set()
  for reference in cve_item["cve"]["references"]["reference_data"]:

vulnerabilities/importers/redhat.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2017 nexB Inc. and others. All rights reserved.
+# Copyright (c) nexB Inc. and others. All rights reserved.
 # http://nexb.com and https://github.com/nexB/vulnerablecode/
 # The VulnerableCode software is licensed under the Apache License version 2.0.
 # Data generated with VulnerableCode require an acknowledgment.
@@ -17,24 +17,27 @@
 # OR CONDITIONS OF ANY KIND, either express or implied. No content created from
 # VulnerableCode should be considered or used as legal advice. Consult an Attorney
 # for any legal advice.
-# VulnerableCode is a free software code scanning tool from nexB Inc. and others.
+# VulnerableCode is a free software code from nexB Inc. and others.
 # Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
-import requests
+
+import json
 
 from packageurl import PackageURL
+import requests
 
 from vulnerabilities.data_source import Advisory
 from vulnerabilities.data_source import DataSource
 from vulnerabilities.data_source import DataSourceConfiguration
 from vulnerabilities.data_source import Reference
+from vulnerabilities.data_source import VulnerabilitySeverity
+from vulnerabilities.severity_systems import scoring_systems
 
 
 class RedhatDataSource(DataSource):
  CONFIG_CLASS = DataSourceConfiguration
 
  def __enter__(self):
-
  self.redhat_response = fetch()
 
  def updated_advisories(self):
@@ -52,7 +55,6 @@ def fetch():
  url = "https://access.redhat.com/hydra/rest/securitydata/cve.json?page={}"
 
  while True:
-
  resp_json = requests.get(url.format(page_no)).json()
  page_no += 1
  if not resp_json:
@@ -65,31 +67,74 @@ def fetch():
 
 
 def to_advisory(advisory_data):
-
  affected_purls = []
  if advisory_data.get("affected_packages"):
  for rpm in advisory_data["affected_packages"]:
  if rpm_to_purl(rpm):
  affected_purls.append(rpm_to_purl(rpm))
 
  references = []
- if advisory_data.get("bugzilla"):
- bugzilla = advisory_data.get("bugzilla")
+ bugzilla = advisory_data.get("bugzilla")
+ if bugzilla:
+ url = "https://bugzilla.redhat.com/show_bug.cgi?id={}".format(bugzilla)
+ bugzilla_data = requests.get(f"https://bugzilla.redhat.com/rest/bug/{bugzilla}").json()
+ bugzilla_severity_val = bugzilla_data["bugs"][0]["severity"]
+ bugzilla_severity = VulnerabilitySeverity(
+ system=scoring_systems["rhbs"],
+ value=bugzilla_severity_val,
+ )
+
  references.append(
  Reference(
- url="https://bugzilla.redhat.com/show_bug.cgi?id={}".format(bugzilla),
+ severities=[bugzilla_severity],
+ url=url,
  reference_id=bugzilla,
  )
  )
 
- for rhsa in advisory_data["advisories"]:
- references.append(
- Reference(
- url="https://access.redhat.com/errata/{}".format(rhsa), reference_id=rhsa,
+ for rh_adv in advisory_data["advisories"]:
+ # RH provides 3 types of advisories RHSA, RHBA, RHEA. Only RHSA's contain severity score.
+ # See https://access.redhat.com/articles/2130961 for more details.
+
+ if "RHSA" in rh_adv.upper():
+ rhsa_data = requests.get(f"https://access.redhat.com/hydra/rest/securitydata/cvrf/{rh_adv}.json").json() # nopep8
+ value = rhsa_data["cvrfdoc"]["aggregate_severity"]
+ rhsa_aggregate_severity = VulnerabilitySeverity(
+ system=scoring_systems["rhas"],
+ value=value,
+ )
+
+ references.append(
+ Reference(
+ severities=[rhsa_aggregate_severity],
+ url="https://access.redhat.com/errata/{}".format(rh_adv),
+ reference_id=rh_adv,
+ )
+ )
+
+ else:
+ references.append(Reference(severities=[], url=url, reference_id=rh_adv))
+
+ redhat_scores = []
+ cvssv3_score = advisory_data.get("cvss3_score")
+ if cvssv3_score:
+ redhat_scores.append(
+ VulnerabilitySeverity(
+ system=scoring_systems["cvssv3"],
+ value=cvssv3_score,
+ )
+ )
+
+ cvssv3_vector = advisory_data.get("cvss3_scoring_vector")
+ if cvssv3_vector:
+ redhat_scores.append(
+ VulnerabilitySeverity(
+ system=scoring_systems["cvssv3_vector"],
+ value=cvssv3_vector,
  )
  )
 
- references.append(Reference(url=advisory_data["resource_url"]))
+ references.append(Reference(severities=redhat_scores, url=advisory_data["resource_url"]))
 
  return Advisory(
  summary=advisory_data["bugzilla_description"],

vulnerabilities/migrations/0001_initial.py

@@ -1,4 +1,4 @@
-# Generated by Django 3.0.7 on 2021-01-21 16:32
+# Generated by Django 3.0.7 on 2021-02-03 07:30
 
 import django.contrib.postgres.fields.jsonb
 from django.db import migrations, models
@@ -49,12 +49,24 @@ class Migration(migrations.Migration):
  ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
  ('cve_id', models.CharField(help_text='CVE ID', max_length=50, null=True, unique=True)),
  ('summary', models.TextField(blank=True, help_text='Summary of the vulnerability')),
- ('cvss', models.FloatField(help_text='CVSS Score', max_length=100, null=True)),
  ],
  options={
  'verbose_name_plural': 'Vulnerabilities',
  },
  ),
+ migrations.CreateModel(
+ name='VulnerabilityReference',
+ fields=[
+ ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+ ('source', models.CharField(blank=True, help_text='Source(s) name eg:NVD', max_length=50)),
+ ('reference_id', models.CharField(blank=True, help_text='Reference ID, eg:DSA-4465-1', max_length=50)),
+ ('url', models.URLField(blank=True, help_text='URL of Vulnerability data', max_length=1024)),
+ ('vulnerability', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='vulnerabilities.Vulnerability')),
+ ],
+ options={
+ 'unique_together': {('vulnerability', 'source', 'reference_id', 'url')},
+ },
+ ),
  migrations.CreateModel(
  name='PackageRelatedVulnerability',
  fields=[
@@ -74,16 +86,16 @@ class Migration(migrations.Migration):
  field=models.ManyToManyField(through='vulnerabilities.PackageRelatedVulnerability', to='vulnerabilities.Vulnerability'),
  ),
  migrations.CreateModel(
- name='VulnerabilityReference',
+ name='VulnerabilitySeverity',
  fields=[
  ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
- ('source', models.CharField(blank=True, help_text='Source(s) name eg:NVD', max_length=50)),
- ('reference_id', models.CharField(blank=True, help_text='Reference ID, eg:DSA-4465-1', max_length=50)),
- ('url', models.URLField(blank=True, help_text='URL of Vulnerability data', max_length=1024)),
+ ('value', models.CharField(help_text='Example: 9.0, Important, High', max_length=50)),
+ ('scoring_system', models.CharField(choices=[('cvssv2', 'CVSSv2'), ('cvssv3', 'CVSSv3'), ('rhbs', 'RedHat Bugzilla severity'), ('rhas', 'RedHat Aggregate severity')], help_text='Identifier for the scoring system used. Avaiable choices are: cvssv2 is identifier for CVSSv2 system, cvssv3 is identifier for CVSSv3 system, rhbs is identifier for RedHat Bugzilla severity system, rhas is identifier for RedHat Aggregate severity system ', max_length=50)),
+ ('reference', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='vulnerabilities.VulnerabilityReference')),
  ('vulnerability', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='vulnerabilities.Vulnerability')),
  ],
  options={
- 'unique_together': {('vulnerability', 'source', 'reference_id', 'url')},
+ 'unique_together': {('vulnerability', 'reference', 'scoring_system')},
  },
  ),
  migrations.AlterUniqueTogether(