Add data migration for old npm and pypa advisory

- Update the created_by field on old advisory to new pipeline_id

Signed-off-by: Keshav Priyadarshi <git@keshav.space>

Author: keshav-space

vulnerabilities/migrations/0063_update_npm_pypa_advisory_created_by.py

@@ -0,0 +1,46 @@
+# Generated by Django 4.2.15 on 2024-09-12 12:56
+
+from django.db import migrations
+
+"""
+Update the created_by field on Advisory from the old qualified_name
+to the new pipeline_id.
+"""
+
+
+def update_created_by(apps, schema_editor):
+    from vulnerabilities.pipelines.npm_importer import NpmImporterPipeline
+    from vulnerabilities.pipelines.pypa_importer import PyPaImporterPipeline
+
+    Advisory = apps.get_model("vulnerabilities", "Advisory")
+    Advisory.objects.filter(created_by="vulnerabilities.importers.npm.NpmImporter").update(
+        created_by=NpmImporterPipeline.pipeline_id
+    )
+    Advisory.objects.filter(created_by="vulnerabilities.importers.pypa.PyPaImporter").update(
+        created_by=PyPaImporterPipeline.pipeline_id
+    )
+
+
+
+def reverse_update_created_by(apps, schema_editor):
+    from vulnerabilities.pipelines.npm_importer import NpmImporterPipeline
+    from vulnerabilities.pipelines.pypa_importer import PyPaImporterPipeline
+
+    Advisory = apps.get_model("vulnerabilities", "Advisory")
+    Advisory.objects.filter(created_by=NpmImporterPipeline.pipeline_id).update(
+        created_by="vulnerabilities.importers.npm.NpmImporter"
+    )
+    Advisory.objects.filter(created_by=PyPaImporterPipeline.pipeline_id).update(
+        created_by="vulnerabilities.importers.pypa.PyPaImporter"
+    )
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0062_package_is_ghost"),
+    ]
+
+    operations = [
+        migrations.RunPython(update_created_by, reverse_code=reverse_update_created_by),
+    ]

vulnerabilities/pipelines/npm_importer.py

@@ -32,6 +32,8 @@
 class NpmImporterPipeline(VulnerableCodeBaseImporterPipeline):
     """Collect advisories from nodejs GitHub repository."""
 
+    pipeline_id = "npm_importer"
+
     spdx_license_expression = "MIT"
     license_url = "https://github.com/nodejs/security-wg/blob/main/LICENSE.md"
     repo_url = "git+https://github.com/nodejs/security-wg"

vulnerabilities/tests/test_changelog.py

@@ -23,7 +23,7 @@ def test_package_changelog():
     pkg, _ = Package.objects.get_or_create_from_purl("pkg:npm/foo@1.0.0")
     assert PackageChangeLog.objects.filter(package=pkg).count() == 0
     adv = Advisory.objects.create(
-        created_by=NpmImporterPipeline.qualified_name,
+        created_by=NpmImporterPipeline.pipeline_id,
         summary="TEST",
         date_collected=datetime.now(),
         url="https://test.com/source",
@@ -49,7 +49,7 @@ def test_package_changelog():
     pkg1, _ = Package.objects.get_or_create_from_purl("pkg:npm/foo@2.0.0")
     assert PackageChangeLog.objects.filter(package=pkg1).count() == 0
     adv = Advisory.objects.create(
-        created_by=NpmImporterPipeline.qualified_name,
+        created_by=NpmImporterPipeline.pipeline_id,
         summary="TEST-1",
         date_collected=datetime.now(),
         url="https://test.com/source-1",
@@ -79,7 +79,7 @@ def test_package_changelog():
 @pytest.mark.django_db
 def test_vulnerability_changelog():
     adv = Advisory.objects.create(
-        created_by=NpmImporterPipeline.qualified_name,
+        created_by=NpmImporterPipeline.pipeline_id,
         summary="TEST_1",
         date_collected=datetime.now(),
         url="https://test.com/source",

vulnerabilities/tests/test_data_migrations.py

@@ -11,8 +11,14 @@
 from django.db import connection
 from django.db.migrations.executor import MigrationExecutor
 from django.test import TestCase
+from django.utils import timezone
+from packageurl import PackageURL
+from univers.version_range import VersionRange
 
 from vulnerabilities import severity_systems
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import Reference
 
 
 class TestMigrations(TestCase):
@@ -610,3 +616,68 @@ def setUpBeforeMigration(self, apps):
     def test_removal_of_duped_purls(self):
         Package = apps.get_model("vulnerabilities", "Package")
         assert Package.objects.count() == 1
+
+
+class TestUpdateNpmPypaAdvisoryCreatedByField(TestMigrations):
+    app_name = "vulnerabilities"
+    migrate_from = "0062_package_is_ghost"
+    migrate_to = "0063_update_npm_pypa_advisory_created_by"
+
+    advisory_data1 = AdvisoryData(
+        aliases=["CVE-2020-13371337"],
+        summary="vulnerability description here",
+        affected_packages=[
+            AffectedPackage(
+                package=PackageURL(type="npm", name="dummy"),
+                affected_version_range=VersionRange.from_string("vers:npm/>=1.0.0|<=2.0.0"),
+            )
+        ],
+        references=[Reference(url="https://example.com/with/more/info/CVE-2020-13371337")],
+        date_published=timezone.now(),
+        url="https://test.com",
+    )
+    advisory_data2 = AdvisoryData(
+        aliases=["CVE-2020-1337"],
+        summary="vulnerability description here",
+        affected_packages=[
+            AffectedPackage(
+                package=PackageURL(type="pypi", name="dummy"),
+                affected_version_range=VersionRange.from_string("vers:pypi/>=1.0.0|<=2.0.0"),
+            )
+        ],
+        references=[Reference(url="https://example.com/with/more/info/CVE-2020-1337")],
+        date_published=timezone.now(),
+        url="https://test2.com",
+    )
+
+    def setUpBeforeMigration(self, apps):
+        Advisory = apps.get_model("vulnerabilities", "Advisory")
+        adv1 = Advisory.objects.create(
+            aliases=self.advisory_data1.aliases,
+            summary=self.advisory_data1.summary,
+            affected_packages=[pkg.to_dict() for pkg in self.advisory_data1.affected_packages],
+            references=[ref.to_dict() for ref in self.advisory_data1.references],
+            url=self.advisory_data1.url,
+            created_by="vulnerabilities.importers.npm.NpmImporter",
+            date_collected=timezone.now(),
+        )
+
+        adv2 = Advisory.objects.create(
+            aliases=self.advisory_data2.aliases,
+            summary=self.advisory_data2.summary,
+            affected_packages=[pkg.to_dict() for pkg in self.advisory_data2.affected_packages],
+            references=[ref.to_dict() for ref in self.advisory_data2.references],
+            url=self.advisory_data2.url,
+            created_by="vulnerabilities.importers.pypa.PyPaImporter",
+            date_collected=timezone.now(),
+        )
+
+    def test_removal_of_duped_purls(self):
+        Advisory = apps.get_model("vulnerabilities", "Advisory")
+        adv = Advisory.objects.all()
+
+        assert adv.filter(created_by="vulnerabilities.importers.pypa.PyPaImporter").count() == 0
+        assert adv.filter(created_by="pypa_importer").count() == 1
+
+        assert adv.filter(created_by="vulnerabilities.importers.npm.NpmImporter").count() == 0
+        assert adv.filter(created_by="npm_importer").count() == 1