[WIP] Drop batch processing and store encountered advisories

vulnerabilities/data_source.py

@@ -40,7 +40,7 @@
 import xml.etree.ElementTree as ET
 
 import pygit2
-from packageurl import PackageURL
+from packageurl import PackageURL, normalize
 from univers.version_specifier import VersionSpecifier
 from univers.versions import version_class_by_package_type
 
@@ -96,8 +96,8 @@ def __post_init__(self):
             raise ValueError("CVE expected, found: {}".format(self.vulnerability_id))
 
     def normalized(self):
-        impacted_package_urls = {package_url for package_url in self.impacted_package_urls}
-        resolved_package_urls = {package_url for package_url in self.resolved_package_urls}
+        impacted_package_urls = sorted([package_url for package_url in self.impacted_package_urls])
+        resolved_package_urls = sorted([package_url for package_url in self.resolved_package_urls])
         references = sorted(
             self.references, key=lambda reference: (reference.reference_id, reference.url)
         )
@@ -112,6 +112,18 @@ def normalized(self):
             references=references,
         )
 
+    def to_dict(self):
+        adv = self.normalized()
+        data = {}
+
+        data["vulnerability_id"] = self.vulnerability_id
+        data["impacted_package_urls"] = [pkg.to_string() for pkg in  self.impacted_package_urls]
+        data["resolved_package_urls"] = [pkg.to_string() for pkg in  self.resolved_package_urls]
+        data["references"] = [dataclasses.asdict(reference) for reference in self.references]
+        data["summary"] = self.summary
+
+        return data
+
 
 class InvalidConfigurationError(Exception):
     pass

vulnerabilities/import_runner.py

@@ -41,22 +41,6 @@
 
 logger = logging.getLogger(__name__)
 
-# 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.
-
-
-@dataclasses.dataclass(frozen=True)
-class PackageRelatedVulnerabilityInserter:
-    vulnerability: models.Vulnerability
-    is_vulnerable: bool
-    package: models.Package
-
-    def to_model_object(self):
-        return models.PackageRelatedVulnerability(**dataclasses.asdict(self))
-
-
 class ImportRunner:
     """
     The ImportRunner is responsible for inserting and updating data about vulnerabilities and
@@ -92,151 +76,53 @@ def run(self, cutoff_date: datetime.datetime = None) -> None:
         logger.info(f"Starting import for {self.importer.name}.")
         data_source = self.importer.make_data_source(self.batch_size, cutoff_date=cutoff_date)
         with data_source:
-            process_advisories(data_source)
+            self.import_run = models.ImportRun.objects.create(importer=self.importer)
+            self.process_advisories(data_source)
+            self.importer.last_import = self.import_run
+
         self.importer.last_run = datetime.datetime.now(tz=datetime.timezone.utc)
         self.importer.data_source_cfg = dataclasses.asdict(data_source.config)
-        self.importer.save()
 
         logger.info(f"Finished import for {self.importer.name}.")
 
-
-def vuln_ref_exists(vulnerability, url, reference_id):
-    return models.VulnerabilityReference.objects.filter(
-        vulnerability=vulnerability, reference_id=reference_id, url=url
-    ).exists()
-
-
-def get_vuln_pkg_refs(vulnerability, package):
-    return models.PackageRelatedVulnerability.objects.filter(
-        vulnerability=vulnerability,
-        package=package,
-    )
-
-
-def process_advisories(data_source: DataSource) -> None:
-    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
-    # single method.
-    advisory_batches = chain(data_source.updated_advisories(), data_source.added_advisories())
-    for batch in advisory_batches:
-        for advisory in batch:
-            try:
-                vuln, vuln_created = _get_or_create_vulnerability(advisory)
-                for vuln_ref in advisory.references:
-                    ref, _ = models.VulnerabilityReference.objects.get_or_create(
-                        vulnerability=vuln, reference_id=vuln_ref.reference_id, url=vuln_ref.url
-                    )
-
-                    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)
-                    is_vulnerable = purl in advisory.impacted_package_urls
-                    pkg_vuln_ref = PackageRelatedVulnerabilityInserter(
-                        vulnerability=vuln, is_vulnerable=is_vulnerable, package=pkg
-                    )
-
-                    if vuln_created or pkg_created:
-                        bulk_create_vuln_pkg_refs.add(pkg_vuln_ref)
-                        # A vulnerability-package relationship does not exist already if either the
-                        # vulnerability or the package is just created.
-
-                    else:
-                        # insert only if it there is no existing vulnerability-package relationship.
-                        existing_ref = get_vuln_pkg_refs(vuln, pkg)
-                        if not existing_ref:
-                            bulk_create_vuln_pkg_refs.add(pkg_vuln_ref)
-                            # A vulnerability-package relationship does not exist already
-                            # if either the vulnerability or the package is just created.
-
-                        else:
-                            # insert only if it there is no existing vulnerability-package relationship.  # nopep8
-                            existing_ref = get_vuln_pkg_refs(vuln, pkg)
-                            if not existing_ref:
-                                bulk_create_vuln_pkg_refs.add(pkg_vuln_ref)
-
-                            else:
-                                # This handles conflicts between existing data and obtained data
-                                if existing_ref[0].is_vulnerable != pkg_vuln_ref.is_vulnerable:
-                                    handle_conflicts(
-                                        [existing_ref[0], pkg_vuln_ref.to_model_object()]
-                                    )
-                                    existing_ref.delete()
-
-            except Exception:
-                # TODO: store error but continue
-                logger.error(
-                    f"Failed to process advisory: {advisory!r}:\n" + traceback.format_exc()
+    def process_advisories(self, data_source: DataSource) -> None:
+        for advisory in data_source.updated_advisories():
+            models.Advisory.objects.create(content=advisory.to_dict(), import_run=self.import_run)
+            if self.importer.last_import and self.importer.last_import.advisory_set.filter(content=advisory.to_dict()):
+                continue
+
+            vuln, _ = _get_or_create_vulnerability(advisory)
+            for packageurl in  advisory.impacted_package_urls:
+                pkg, _ = _get_or_create_package(packageurl)
+                models.PackageRelatedVulnerability.objects.get_or_create(
+                    package=pkg,
+                    is_vulnerable=True,
+                    vulnerability=vuln
+                )
+
+            for packageurl in  advisory.resolved_package_urls:
+                pkg, _ = _get_or_create_package(packageurl)
+                models.PackageRelatedVulnerability.objects.get_or_create(
+                    package=pkg,
+                    is_vulnerable=False,
+                    vulnerability=vuln
+                )
+
+            for reference in advisory.references:
+                models.VulnerabilityReference.objects.get_or_create(
+                    vulnerability=vuln,
+                    reference_id=reference.reference_id,
+                    url=reference.url
                 )
-
-    # find_conflicting_relations handles in-memory conflicts
-    conflicts = find_conflicting_relations(bulk_create_vuln_pkg_refs)
-
-    models.PackageRelatedVulnerability.objects.bulk_create(
-        [i.to_model_object() for i in bulk_create_vuln_pkg_refs if i not in conflicts]
-    )
-
-    handle_conflicts([i.to_model_object() for i in conflicts])
-
-
-def find_conflicting_relations(
-    relations: Set[Set[PackageRelatedVulnerabilityInserter]],
-) -> Set[PackageRelatedVulnerabilityInserter]:
-
-    # Chop off `is_vulnerable` flag from PackageRelatedVulnerabilityInserter and create a list of
-    # tuples of format (rel.package, rel.vulnerability)
-
-    relation_tuples = [(rel.package, rel.vulnerability) for rel in relations]
-    relation_counter = Counter(relation_tuples).most_common()
-
-    # If a (rel.package, rel.vulnerability) occurs twice then that means the
-    # PackageRelatedVulnerabilityInserter objects
-    # (rel.package, rel.vulnerability, is_vulnerable=True) and
-    # (rel.package, rel.vulnerability, is_vulnerable=False) both existed which is conflicting data.
-    # We detect and return these conflicts.
-
-    conflicts = set()
-    for rel, count in relation_counter:
-        if count < 2:
-            # All the subsequent entries from here on would have count == 1 which is of no interest
-            # since conflicts exist in pairs with `is_vulnerable=True` and `is_vulnerable=False`.
-            break
-
-        # `rel` is of format (pkg, vuln)
-        conflicts.add(
-            PackageRelatedVulnerabilityInserter(
-                vulnerability=rel[1], package=rel[0], is_vulnerable=True
-            )
-        )
-
-        conflicts.add(
-            PackageRelatedVulnerabilityInserter(
-                vulnerability=rel[1], package=rel[0], is_vulnerable=False
-            )
-        )
-
-    return conflicts
-
-
-def handle_conflicts(conflicts):
-    conflicts = serializers.serialize("json", [i for i in conflicts])
-    models.ImportProblem.objects.create(conflicting_model=conflicts)
-
 
 def _get_or_create_vulnerability(
     advisory: Advisory,
 ) -> Tuple[models.Vulnerability, bool]:
 
     vuln, created = models.Vulnerability.objects.get_or_create(
         vulnerability_id=advisory.vulnerability_id
-    )  # nopep8
+    )
+
     # Eventually we only want to keep summary from NVD and ignore other descriptions.
     if advisory.summary and vuln.summary != advisory.summary:
         vuln.summary = advisory.summary
@@ -258,9 +144,3 @@ def _get_or_create_package(p: PackageURL) -> Tuple[models.Package, bool]:
             query_kwargs[key] = val
 
     return models.Package.objects.get_or_create(**query_kwargs)
-
-
-def _package_url_to_package(purl: PackageURL) -> models.Package:
-    p = models.Package()
-    p.set_package_url(purl)
-    return p

vulnerabilities/importers/openssl.py

@@ -50,19 +50,16 @@ def updated_advisories(self) -> Set[Advisory]:
         # (url, etag) mappings in the DB. `create_etag`  creates
         # (url, etag) pair. If a (url, etag) already exists then the code
         # skips processing the response further to avoid duplicate work
-        if create_etag(data_src=self, url=self.url, etag_key="ETag"):
-            raw_data = self.fetch()
-            advisories = self.to_advisories(raw_data)
-            return self.batch_advisories(advisories)
+        raw_data = self.fetch()
+        for advisory in self.to_advisories(raw_data):
+            yield advisory
 
-        return []
 
     def fetch(self):
         return requests.get(self.url).content
 
     @staticmethod
     def to_advisories(xml_response: str) -> Set[Advisory]:
-        advisories = []
         pkg_name = "openssl"
         pkg_type = "generic"
         root = ET.fromstring(xml_response)
@@ -110,13 +107,10 @@ def to_advisories(xml_response: str) -> Set[Advisory]:
                     for version in vuln_pkg_versions
                 }
 
-                advisory = Advisory(
+                yield Advisory(
                     vulnerability_id=cve_id,
                     summary=summary,
                     impacted_package_urls=vuln_purls,
                     resolved_package_urls=safe_purls,
                     references=ref_urls,
                 )
-                advisories.append(advisory)
-
-        return advisories

vulnerabilities/migrations/0002_auto_20210420_0559.py

@@ -0,0 +1,51 @@
+# Generated by Django 3.0.14 on 2021-04-20 05:59
+
+import django.contrib.postgres.fields.jsonb
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('vulnerabilities', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name='importer',
+            name='last_run',
+        ),
+        migrations.AlterField(
+            model_name='vulnerability',
+            name='vulnerability_id',
+            field=models.CharField(help_text="Unique identifier for a vulnerability: this is either a published CVE id (as in CVE-2020-7965) if it exists. Otherwise this is a VulnerableCode-assigned VULCOID (as in VULCOID-20210222-1315-16461541). When a vulnerability CVE is assigned later we replace this with the CVE and keep the 'old' VULCOID in the 'old_vulnerability_id' field to support redirection to the CVE id.", max_length=50, unique=True),
+        ),
+        migrations.AlterField(
+            model_name='vulnerabilityseverity',
+            name='scoring_system',
+            field=models.CharField(choices=[('cvssv2', 'CVSSv2 Base Score'), ('cvssv2_vector', 'CVSSv2 Vector'), ('cvssv3', 'CVSSv3 Base Score'), ('cvssv3_vector', 'CVSSv3 Vector'), ('cvssv3.1', 'CVSSv3.1 Base Score'), ('cvssv3.1_vector', 'CVSSv3.1 Vector'), ('rhbs', 'RedHat Bugzilla severity'), ('rhas', 'RedHat Aggregate severity'), ('avgs', 'Archlinux Vulnerability Group Severity'), ('cvssv3.1_qr', 'CVSSv3.1 Qualitative Severity Rating'), ('generic_textual', 'Generic textual severity rating')], help_text='identifier for the scoring system used. Available choices are: cvssv2 is vulnerability_id for CVSSv2 Base Score system, cvssv2_vector is vulnerability_id for CVSSv2 Vector system, cvssv3 is vulnerability_id for CVSSv3 Base Score system, cvssv3_vector is vulnerability_id for CVSSv3 Vector system, cvssv3.1 is vulnerability_id for CVSSv3.1 Base Score system, cvssv3.1_vector is vulnerability_id for CVSSv3.1 Vector system, rhbs is vulnerability_id for RedHat Bugzilla severity system, rhas is vulnerability_id for RedHat Aggregate severity system, avgs is vulnerability_id for Archlinux Vulnerability Group Severity system, cvssv3.1_qr is vulnerability_id for CVSSv3.1 Qualitative Severity Rating system, generic_textual is vulnerability_id for Generic textual severity rating system ', max_length=50),
+        ),
+        migrations.CreateModel(
+            name='ImportRun',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('date', models.DateField(auto_now=True)),
+                ('importer', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='importer', to='vulnerabilities.Importer')),
+            ],
+        ),
+        migrations.CreateModel(
+            name='Advisory',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('content', django.contrib.postgres.fields.jsonb.JSONField()),
+                ('import_run', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='vulnerabilities.ImportRun')),
+            ],
+        ),
+        migrations.AddField(
+            model_name='importer',
+            name='last_import',
+            field=models.ForeignKey(default=None, on_delete=django.db.models.deletion.CASCADE, related_name='last_import_run', to='vulnerabilities.ImportRun'),
+            preserve_default=False,
+        ),
+    ]

vulnerabilities/migrations/0003_auto_20210420_0622.py

@@ -0,0 +1,19 @@
+# Generated by Django 3.0.14 on 2021-04-20 06:22
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('vulnerabilities', '0002_auto_20210420_0559'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='importer',
+            name='last_import',
+            field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='last_import_run', to='vulnerabilities.ImportRun'),
+        ),
+    ]

vulnerabilities/migrations/0004_importer_last_run.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.0.14 on 2021-04-20 07:51
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('vulnerabilities', '0003_auto_20210420_0622'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='importer',
+            name='last_run',
+            field=models.DateTimeField(help_text='UTC Timestamp of the last run', null=True),
+        ),
+    ]