Merge pull request #1087 from TG1999/fix/1086

Add migrations to remove corrupted advisories #1086

vulnerabilities/migrations/0038_remove_corrupted_advisories_with_incorrect_refs_and_severity.py

@@ -0,0 +1,36 @@
+#
+# 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
+
+from vulnerabilities.severity_systems import SCORING_SYSTEMS
+
+class Migration(migrations.Migration):
+
+    def remove_advisories(apps, schema_editor):
+        Advisory = apps.get_model("vulnerabilities", "Advisory")
+        deletables = []
+        for advisory in Advisory.objects.iterator(chunk_size=1000):
+            for ref in advisory.references:
+                if not ref["url"]:
+                    deletables.append(advisory.pk)
+                    break
+                for sev in ref["severities"]:
+                    if sev["system"] not in SCORING_SYSTEMS:
+                        deletables.append(advisory.pk)
+                        break
+        Advisory.objects.filter(pk__in=deletables).delete()            
+
+    dependencies = [
+        ("vulnerabilities", "0037_advisory_weaknesses_weakness"),
+    ]
+
+    operations = [
+        migrations.RunPython(remove_advisories, reverse_code=migrations.RunPython.noop),
+    ]

vulnerabilities/tests/test_data_migrations.py

@@ -533,3 +533,38 @@ def test_merge_rows(self):
         ]
 
         assert severities == expected
+
+
+class RemoveCorrupteAdvisories(TestMigrations):
+    app_name = "vulnerabilities"
+    migrate_from = "0037_advisory_weaknesses_weakness"
+    migrate_to = "0038_remove_corrupted_advisories_with_incorrect_refs_and_severity"
+
+    def setUpBeforeMigration(self, apps):
+        # using get_model to avoid circular import
+        Advisory = apps.get_model("vulnerabilities", "Advisory")
+
+        corrupted_advisory = Advisory.objects.create(
+            aliases=["CVE-2020-1234"],
+            summary="Corrupted advisory",
+            references=[
+                {
+                    "reference_id": "cpe:2.3:a:f5:nginx:1.16.1:*:*:*:*:*:*:*",
+                    "url": "",
+                    "severity": [
+                        {
+                            "scoring_system": "cvssv3_vector",
+                            "value": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H",
+                        }
+                    ],
+                }
+            ],
+            date_collected="2020-01-01",
+            date_published="2020-01-01",
+        )
+        corrupted_advisory.save()
+
+    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