Do not create vulnerabilities for empty aliases (#1334)

* Do not create vulnerabilities for empty aliases

Signed-off-by: Tushar Goel <[email protected]>

* Add test for process_inferences

Signed-off-by: Tushar Goel <[email protected]>

* Refactor get_or_create_vulnerability_and_aliases function

Signed-off-by: Tushar Goel <[email protected]>

* Copy changes of improver runner in import runner

Signed-off-by: Tushar Goel <[email protected]>

* Add migration to remove vulnerabilities with empty aliases

Signed-off-by: Tushar Goel <[email protected]>

* Add tests for migration

Signed-off-by: Tushar Goel <[email protected]>

* Address review comments

Signed-off-by: Tushar Goel <[email protected]>

---------

Signed-off-by: Tushar Goel <[email protected]>

Author: TG1999

vulnerabilities/import_runner.py

@@ -245,69 +245,97 @@ def create_valid_vulnerability_reference(url, reference_id=None):
     return reference
 
 
-def get_or_create_vulnerability_and_aliases(alias_names, vulnerability_id=None, summary=None):
+def get_or_create_vulnerability_and_aliases(
+    aliases: List[str], vulnerability_id=None, summary=None
+):
     """
     Get or create vulnerabilitiy and aliases such that all existing and new
     aliases point to the same vulnerability
     """
-    existing_vulns = set()
-    alias_names = set(alias_names)
-    new_alias_names = set()
-    for alias_name in alias_names:
-        try:
-            alias = Alias.objects.get(alias=alias_name)
-            existing_vulns.add(alias.vulnerability)
-        except Alias.DoesNotExist:
-            new_alias_names.add(alias_name)
-
-    # If given set of aliases point to different vulnerabilities in the
-    # database, request is malformed
-    # TODO: It is possible that all those vulnerabilities are actually
-    # the same at data level, figure out a way to merge them
-    if len(existing_vulns) > 1:
-        logger.warning(
-            f"Given aliases {alias_names} already exist and do not point "
-            f"to a single vulnerability. Cannot improve. Skipped."
-        )
-        return
+    aliases = set(alias.strip() for alias in aliases if alias and alias.strip())
+    new_alias_names, existing_vulns = get_vulns_for_aliases_and_get_new_aliases(aliases)
+
+    # All aliases must point to the same vulnerability
+    vulnerability = None
+    if existing_vulns:
+        if len(existing_vulns) != 1:
+            vcids = ", ".join(v.vulnerability_id for v in existing_vulns)
+            logger.error(
+                f"Cannot create vulnerability. "
+                f"Aliases {aliases} already exist and point "
+                f"to multiple vulnerabilities {vcids}."
+            )
+            return
+        else:
+            vulnerability = existing_vulns.pop()
 
-    existing_alias_vuln = existing_vulns.pop() if existing_vulns else None
-
-    if (
-        existing_alias_vuln
-        and vulnerability_id
-        and existing_alias_vuln.vulnerability_id != vulnerability_id
-    ):
-        logger.warning(
-            f"Given aliases {alias_names!r} already exist and point to existing"
-            f"vulnerability {existing_alias_vuln}. Unable to create Vulnerability "
-            f"with vulnerability_id {vulnerability_id}. Skipped"
-        )
-        return
+            if vulnerability_id and vulnerability.vulnerability_id != vulnerability_id:
+                logger.error(
+                    f"Cannot create vulnerability. "
+                    f"Aliases {aliases} already exist and point to a different "
+                    f"vulnerability {vulnerability} than the requested "
+                    f"vulnerability {vulnerability_id}."
+                )
+                return
 
-    if existing_alias_vuln:
-        vulnerability = existing_alias_vuln
-    elif vulnerability_id:
+    if vulnerability_id and not vulnerability:
         try:
             vulnerability = Vulnerability.objects.get(vulnerability_id=vulnerability_id)
         except Vulnerability.DoesNotExist:
-            logger.warning(
-                f"Given vulnerability_id: {vulnerability_id} does not exist in the database"
-            )
+            logger.error(f"Cannot get requested vulnerability {vulnerability_id}.")
             return
+    if vulnerability:
+        # TODO: We should keep multiple summaries, one for each advisory
+        # if summary and summary != vulnerability.summary:
+        #     logger.warning(
+        #         f"Inconsistent summary for {vulnerability.vulnerability_id}. "
+        #         f"Existing: {vulnerability.summary!r}, provided: {summary!r}"
+        #     )
+        associate_vulnerability_with_aliases(vulnerability=vulnerability, aliases=new_alias_names)
     else:
-        vulnerability = Vulnerability(summary=summary)
-        vulnerability.save()
+        try:
+            vulnerability = create_vulnerability_and_add_aliases(
+                aliases=new_alias_names, summary=summary
+            )
+        except Exception as e:
+            logger.error(
+                f"Cannot create vulnerability with summary {summary!r} and {new_alias_names!r} {e!r}.\n{traceback_format_exc()}."
+            )
+            return
+
+    return vulnerability
+
+
+def get_vulns_for_aliases_and_get_new_aliases(aliases):
+    """
+    Return ``new_aliases`` that are not in the database and
+    ``existing_vulns`` that point to the given ``aliases``.
+    """
+    new_aliases = set(aliases)
+    existing_vulns = set()
+    for alias in Alias.objects.filter(alias__in=aliases):
+        existing_vulns.add(alias.vulnerability)
+        new_aliases.remove(alias.alias)
+    return new_aliases, existing_vulns
 
-    if summary and summary != vulnerability.summary:
-        logger.warning(
-            f"Inconsistent summary for {vulnerability!r}. "
-            f"Existing: {vulnerability.summary}, provided: {summary}"
-        )
 
-    for alias_name in new_alias_names:
+@transaction.atomic
+def create_vulnerability_and_add_aliases(aliases, summary):
+    """
+    Return a new ``vulnerability`` created with ``summary``
+    and associate the ``vulnerability`` with ``aliases``.
+    Raise exception if no alias is associated with the ``vulnerability``.
+    """
+    vulnerability = Vulnerability(summary=summary)
+    vulnerability.save()
+    associate_vulnerability_with_aliases(aliases, vulnerability)
+    if not vulnerability.aliases.count():
+        raise Exception(f"Vulnerability {vulnerability.vcid} must have one or more aliases")
+    return vulnerability
+
+
+def associate_vulnerability_with_aliases(aliases, vulnerability):
+    for alias_name in aliases:
         alias = Alias(alias=alias_name, vulnerability=vulnerability)
         alias.save()
         logger.info(f"New alias for {vulnerability!r}: {alias_name}")
-
-    return vulnerability

vulnerabilities/improve_runner.py

@@ -10,6 +10,7 @@
 import logging
 from datetime import datetime
 from datetime import timezone
+from traceback import format_exc as traceback_format_exc
 from typing import List
 
 from django.core.exceptions import ValidationError
@@ -79,7 +80,7 @@ def process_inferences(inferences: List[Inference], advisory: Advisory, improver
     for inference in inferences:
         vulnerability = get_or_create_vulnerability_and_aliases(
             vulnerability_id=inference.vulnerability_id,
-            alias_names=inference.aliases,
+            aliases=inference.aliases,
             summary=inference.summary,
         )
 
@@ -173,69 +174,97 @@ def create_valid_vulnerability_reference(url, reference_id=None):
     return reference
 
 
-def get_or_create_vulnerability_and_aliases(alias_names, vulnerability_id=None, summary=None):
+def get_or_create_vulnerability_and_aliases(
+    aliases: List[str], vulnerability_id=None, summary=None
+):
     """
     Get or create vulnerabilitiy and aliases such that all existing and new
     aliases point to the same vulnerability
     """
-    existing_vulns = set()
-    alias_names = set(alias_names)
-    new_alias_names = set()
-    for alias_name in alias_names:
-        try:
-            alias = Alias.objects.get(alias=alias_name)
-            existing_vulns.add(alias.vulnerability)
-        except Alias.DoesNotExist:
-            new_alias_names.add(alias_name)
-
-    # If given set of aliases point to different vulnerabilities in the
-    # database, request is malformed
-    # TODO: It is possible that all those vulnerabilities are actually
-    # the same at data level, figure out a way to merge them
-    if len(existing_vulns) > 1:
-        logger.warning(
-            f"Given aliases {alias_names} already exist and do not point "
-            f"to a single vulnerability. Cannot improve. Skipped."
-        )
-        return
-
-    existing_alias_vuln = existing_vulns.pop() if existing_vulns else None
-
-    if (
-        existing_alias_vuln
-        and vulnerability_id
-        and existing_alias_vuln.vulnerability_id != vulnerability_id
-    ):
-        logger.warning(
-            f"Given aliases {alias_names!r} already exist and point to existing"
-            f"vulnerability {existing_alias_vuln}. Unable to create Vulnerability "
-            f"with vulnerability_id {vulnerability_id}. Skipped"
-        )
-        return
+    aliases = set(alias.strip() for alias in aliases if alias and alias.strip())
+    new_alias_names, existing_vulns = get_vulns_for_aliases_and_get_new_aliases(aliases)
+
+    # All aliases must point to the same vulnerability
+    vulnerability = None
+    if existing_vulns:
+        if len(existing_vulns) != 1:
+            vcids = ", ".join(v.vulnerability_id for v in existing_vulns)
+            logger.error(
+                f"Cannot create vulnerability. "
+                f"Aliases {aliases} already exist and point "
+                f"to multiple vulnerabilities {vcids}."
+            )
+            return
+        else:
+            vulnerability = existing_vulns.pop()
+
+            if vulnerability_id and vulnerability.vulnerability_id != vulnerability_id:
+                logger.error(
+                    f"Cannot create vulnerability. "
+                    f"Aliases {aliases} already exist and point to a different "
+                    f"vulnerability {vulnerability} than the requested "
+                    f"vulnerability {vulnerability_id}."
+                )
+                return
 
-    if existing_alias_vuln:
-        vulnerability = existing_alias_vuln
-    elif vulnerability_id:
+    if vulnerability_id and not vulnerability:
         try:
             vulnerability = Vulnerability.objects.get(vulnerability_id=vulnerability_id)
         except Vulnerability.DoesNotExist:
-            logger.warning(
-                f"Given vulnerability_id: {vulnerability_id} does not exist in the database"
-            )
+            logger.error(f"Cannot get requested vulnerability {vulnerability_id}.")
             return
+    if vulnerability:
+        # TODO: We should keep multiple summaries, one for each advisory
+        # if summary and summary != vulnerability.summary:
+        #     logger.warning(
+        #         f"Inconsistent summary for {vulnerability.vulnerability_id}. "
+        #         f"Existing: {vulnerability.summary!r}, provided: {summary!r}"
+        #     )
+        associate_vulnerability_with_aliases(vulnerability=vulnerability, aliases=new_alias_names)
     else:
-        vulnerability = Vulnerability(summary=summary)
-        vulnerability.save()
+        try:
+            vulnerability = create_vulnerability_and_add_aliases(
+                aliases=new_alias_names, summary=summary
+            )
+        except Exception as e:
+            logger.error(
+                f"Cannot create vulnerability with summary {summary!r} and {new_alias_names!r} {e!r}.\n{traceback_format_exc()}."
+            )
+            return
 
-    if summary and summary != vulnerability.summary:
-        logger.warning(
-            f"Inconsistent summary for {vulnerability!r}. "
-            f"Existing: {vulnerability.summary}, provided: {summary}"
-        )
+    return vulnerability
+
+
+def get_vulns_for_aliases_and_get_new_aliases(aliases):
+    """
+    Return ``new_aliases`` that are not in the database and
+    ``existing_vulns`` that point to the given ``aliases``.
+    """
+    new_aliases = set(aliases)
+    existing_vulns = set()
+    for alias in Alias.objects.filter(alias__in=aliases):
+        existing_vulns.add(alias.vulnerability)
+        new_aliases.remove(alias.alias)
+    return new_aliases, existing_vulns
 
-    for alias_name in new_alias_names:
+
+@transaction.atomic
+def create_vulnerability_and_add_aliases(aliases, summary):
+    """
+    Return a new ``vulnerability`` created with ``summary``
+    and associate the ``vulnerability`` with ``aliases``.
+    Raise exception if no alias is associated with the ``vulnerability``.
+    """
+    vulnerability = Vulnerability(summary=summary)
+    vulnerability.save()
+    associate_vulnerability_with_aliases(aliases, vulnerability)
+    if not vulnerability.aliases.count():
+        raise Exception(f"Vulnerability {vulnerability.vcid} must have one or more aliases")
+    return vulnerability
+
+
+def associate_vulnerability_with_aliases(aliases, vulnerability):
+    for alias_name in aliases:
         alias = Alias(alias=alias_name, vulnerability=vulnerability)
         alias.save()
         logger.info(f"New alias for {vulnerability!r}: {alias_name}")
-
-    return vulnerability

vulnerabilities/migrations/0041_remove_vulns_with_empty_aliases.py

@@ -0,0 +1,37 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/nexB/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0040_remove_advisory_date_improved_advisory_date_imported"),
+    ]
+
+    def remove_vulns_with_empty_aliases(apps, _):
+        Vulnerability = apps.get_model("vulnerabilities", "Vulnerability")
+        Package = apps.get_model("vulnerabilities", "Package")
+        packages = []
+        vulnerabilities = []
+        for vuln in Vulnerability.objects.filter(aliases=None).prefetch_related(
+                "packages"
+        ):
+            # Delete packages associated with that vulnerability
+            for package in vuln.packages.all():
+                packages.append(package.id)
+            vulnerabilities.append(vuln.id)
+
+        Vulnerability.objects.filter(id__in=vulnerabilities).delete()
+        Package.objects.filter(id__in=packages).delete()
+
+    operations = [
+        migrations.RunPython(remove_vulns_with_empty_aliases, reverse_code=migrations.RunPython.noop),
+    ]

vulnerabilities/models.py

@@ -11,6 +11,7 @@
 import json
 import logging
 from contextlib import suppress
+from typing import Any
 
 from cwe2.database import Database
 from django.contrib.auth import get_user_model
@@ -759,6 +760,8 @@ class Alias(models.Model):
     alias = models.CharField(
         max_length=50,
         unique=True,
+        blank=False,
+        null=False,
         help_text="An alias is a unique vulnerability identifier in some database, "
         "such as CVE-2020-2233",
     )

vulnerabilities/tests/test_data_migrations.py

@@ -568,3 +568,29 @@ def test_removal_of_corrupted_advisory(self):
         # using get_model to avoid circular import
         Advisory = self.apps.get_model("vulnerabilities", "Advisory")
         Advisory.objects.all().count() == 0
+
+
+class RemoveVulnerabilitiesWithEmptyAliases(TestMigrations):
+    app_name = "vulnerabilities"
+    migrate_from = "0040_remove_advisory_date_improved_advisory_date_imported"
+    migrate_to = "0041_remove_vulns_with_empty_aliases"
+
+    def setUpBeforeMigration(self, apps):
+        # using get_model to avoid circular import
+        Vulnerability = apps.get_model("vulnerabilities", "Vulnerability")
+        Alias = apps.get_model("vulnerabilities", "Alias")
+
+        vuln = Vulnerability.objects.create(
+            summary="Corrupted vuln",
+        )
+        vuln.save()
+        vuln = Vulnerability.objects.create(
+            summary="vuln",
+        )
+        vuln.save()
+        Alias.objects.create(alias="CVE-123", vulnerability=vuln)
+
+    def test_removal_of_corrupted_vulns(self):
+        # using get_model to avoid circular import
+        Vulnerability = self.apps.get_model("vulnerabilities", "Vulnerability")
+        Vulnerability.objects.all().count() == 1