Merge pull request #2268 from aboutcode-org/store-latest-advisory

feat: add new field to track latest version of v2 advisory

Author: keshav-space

vulnerabilities/migrations/0121_advisoryv2_is_latest_alter_advisoryv2_advisory_id_and_more.py

@@ -0,0 +1,65 @@
+# Generated by Django 5.2.11 on 2026-04-13 19:05
+
+from django.db import migrations
+from django.db import models
+from django.db.models import F
+
+
+class Migration(migrations.Migration):
+    def add_is_latest_on_existing_advisory(apps, schema_editor):
+        Advisory = apps.get_model("vulnerabilities", "AdvisoryV2")
+
+        print(f"\nUpdating is_latest on existing V2 Advisory.")
+        latest_qs = Advisory.objects.order_by(
+            "avid",
+            F("date_collected").desc(nulls_last=True),
+            "-id",
+        ).distinct("avid")
+
+        Advisory.objects.filter(id__in=latest_qs.values("id")).update(is_latest=True)
+
+    dependencies = [
+        ("vulnerabilities", "0120_impactedpackage_last_range_unfurl_at_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="advisoryv2",
+            name="is_latest",
+            field=models.BooleanField(
+                db_index=True,
+                default=False,
+                help_text="Indicates whether this is the latest version of the advisory identified by its AVID.",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="advisoryv2",
+            name="advisory_id",
+            field=models.CharField(
+                db_index=True,
+                help_text="An advisory is a unique vulnerability identifier in some database, such as PYSEC-2020-2233",
+                max_length=500,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="advisoryv2",
+            name="datasource_id",
+            field=models.CharField(
+                db_index=True,
+                help_text="Unique ID for the datasource used for this advisory .e.g.: nginx_importer_v2",
+                max_length=200,
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="advisoryv2",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(("is_latest", True)),
+                fields=("avid",),
+                name="unique_latest_per_avid",
+            ),
+        ),
+        migrations.RunPython(
+            code=add_is_latest_on_existing_advisory,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

vulnerabilities/models.py

@@ -2886,11 +2886,7 @@ def latest_for_avid(self, avid: str):
         )
 
     def latest_per_avid(self):
-        return self.order_by(
-            "avid",
-            F("date_collected").desc(nulls_last=True),
-            "-id",
-        ).distinct("avid")
+        return self.filter(is_latest=True)
 
     def latest_for_avids(self, avids):
         return self.filter(avid__in=avids).latest_per_avid()
@@ -3007,6 +3003,7 @@ class AdvisoryV2(models.Model):
         max_length=200,
         blank=False,
         null=False,
+        db_index=True,
         help_text="Unique ID for the datasource used for this advisory ." "e.g.: nginx_importer_v2",
     )
 
@@ -3016,6 +3013,7 @@ class AdvisoryV2(models.Model):
         blank=False,
         null=False,
         unique=False,
+        db_index=True,
         help_text="An advisory is a unique vulnerability identifier in some database, "
         "such as PYSEC-2020-2233",
     )
@@ -3090,6 +3088,14 @@ class AdvisoryV2(models.Model):
         help_text="UTC Date on which the advisory was collected",
     )
 
+    is_latest = models.BooleanField(
+        default=False,
+        blank=False,
+        null=False,
+        db_index=True,
+        help_text="Indicates whether this is the latest version of the advisory identified by its AVID.",
+    )
+
     original_advisory_text = models.TextField(
         blank=True,
         null=True,
@@ -3142,6 +3148,11 @@ class AdvisoryV2(models.Model):
     class Meta:
         unique_together = ["datasource_id", "advisory_id", "unique_content_id"]
         ordering = ["datasource_id", "advisory_id", "date_published", "unique_content_id"]
+        constraints = [
+            models.UniqueConstraint(
+                fields=["avid"], condition=Q(is_latest=True), name="unique_latest_per_avid"
+            )
+        ]
         indexes = [
             models.Index(
                 fields=["avid", "-date_collected", "-id"],

vulnerabilities/pipes/advisory.py

@@ -334,6 +334,13 @@ def insert_advisory_v2(
     if not created:
         return advisory_obj
 
+    AdvisoryV2.objects.filter(
+        avid=f"{pipeline_id}/{advisory.advisory_id}",
+        is_latest=True,
+    ).update(is_latest=False)
+    advisory_obj.is_latest = True
+    advisory_obj.save()
+
     aliases = get_or_create_advisory_aliases(aliases=advisory.aliases)
     references = get_or_create_advisory_references(references=advisory.references)
     severities = get_or_create_advisory_severities(severities=advisory.severities)

vulnerabilities/tests/pipelines/v2_improvers/test_collect_ssvc_trees.py

@@ -36,6 +36,7 @@ def vulnrichment_advisory(db):
         url="https://example.com/advisory/TEST-2024-0001",
         unique_content_id="unique-1234",
         date_collected=datetime.now(),
+        is_latest=True,
     )
 
 
@@ -59,6 +60,7 @@ def related_advisory(db):
         url="https://example.com/related/TEST-2024-0001",
         unique_content_id="unique-5678",
         date_collected=datetime.now(),
+        is_latest=True,
     )
 
 

vulnerabilities/tests/pipelines/v2_improvers/test_compute_package_risk_v2.py

@@ -34,6 +34,7 @@ def test_simple_risk_pipeline():
         unique_content_id="ajkef",
         url="https://test.com",
         date_collected=datetime.now(),
+        is_latest=True,
     )
     adv.save()
 

vulnerabilities/tests/pipelines/v2_improvers/test_relate_severities.py

@@ -25,6 +25,7 @@ def test_relate_severities_by_advisory_id():
         unique_content_id="ab1",
         url="https://example.com/advisory/CVE-2024-0001",
         date_collected="2024-01-01",
+        is_latest=True,
     )
 
     severity_advisory = AdvisoryV2.objects.create(
@@ -34,6 +35,7 @@ def test_relate_severities_by_advisory_id():
         unique_content_id="ab2",
         url="https://example.com/epss/CVE-2024-0001",
         date_collected="2024-01-02",
+        is_latest=True,
     )
     severity_advisory.severities.create(
         scoring_system=EPSS.identifier,
@@ -59,6 +61,7 @@ def test_relate_severities_via_alias():
         unique_content_id="ab3",
         url="https://example.com/advisory/CVE-2024-0002",
         date_collected="2024-01-01",
+        is_latest=True,
     )
 
     base.aliases.create(alias="CVE-2024-ALIAS")
@@ -70,6 +73,7 @@ def test_relate_severities_via_alias():
         unique_content_id="ab4",
         url="https://example.com/epss/CVE-2024-ALIAS",
         date_collected="2024-01-02",
+        is_latest=True,
     )
     severity_advisory.severities.create(
         scoring_system=EPSS.identifier,
@@ -91,6 +95,7 @@ def test_no_self_relation_created():
         url="https://example.com/advisory/CVE-2024-0003",
         date_collected="2024-01-03",
         avid="epss/CVE-2024-0003",
+        is_latest=True,
     )
     advisory.severities.create(
         scoring_system=EPSS.identifier,
@@ -112,6 +117,7 @@ def test_unsupported_severity_system_is_ignored():
         url="https://example.com/advisory/CVE-2024-0004",
         date_collected="2024-01-01",
         avid="nvd/CVE-2024-0004",
+        is_latest=True,
     )
 
     severity_advisory = AdvisoryV2.objects.create(
@@ -121,6 +127,7 @@ def test_unsupported_severity_system_is_ignored():
         url="https://example.com/epss/CVE-2024-0004",
         date_collected="2024-01-02",
         avid="epss/CVE-2024-0004",
+        is_latest=True,
     )
     severity_advisory.severities.create(
         scoring_system="UNKNOWN_SYSTEM",
@@ -142,6 +149,7 @@ def test_pipeline_is_idempotent():
         url="https://example.com/advisory/CVE-2024-0005",
         date_collected="2024-01-01",
         avid="nvd/CVE-2024-0005",
+        is_latest=True,
     )
 
     severity = AdvisoryV2.objects.create(
@@ -150,6 +158,7 @@ def test_pipeline_is_idempotent():
         unique_content_id="ab9",
         url="https://example.com/epss/CVE-2024-0005",
         date_collected="2024-01-02",
+        is_latest=True,
         avid="epss/CVE-2024-0005",
     )
     severity.severities.create(

vulnerabilities/tests/pipes/test_advisory.py

@@ -18,12 +18,15 @@
 
 from vulnerabilities import models
 from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.importer import AdvisoryDataV2
 from vulnerabilities.importer import AffectedPackage
+from vulnerabilities.importer import AffectedPackageV2
 from vulnerabilities.importer import PackageCommitPatchData
 from vulnerabilities.importer import Reference
 from vulnerabilities.models import AdvisoryAlias
 from vulnerabilities.models import AdvisoryReference
 from vulnerabilities.models import AdvisorySeverity
+from vulnerabilities.models import AdvisoryV2
 from vulnerabilities.models import AdvisoryWeakness
 from vulnerabilities.models import PackageCommitPatch
 from vulnerabilities.pipes.advisory import get_or_create_advisory_aliases
@@ -33,6 +36,8 @@
 from vulnerabilities.pipes.advisory import get_or_create_advisory_weaknesses
 from vulnerabilities.pipes.advisory import get_or_create_aliases
 from vulnerabilities.pipes.advisory import import_advisory
+from vulnerabilities.pipes.advisory import insert_advisory_v2
+from vulnerabilities.tests.pipelines import TestLogger
 from vulnerabilities.utils import compute_content_id
 
 
@@ -257,3 +262,76 @@ def test_get_or_create_advisory_commit(advisory_commit):
         assert isinstance(commit, PackageCommitPatch)
         assert commit.commit_hash in [c.commit_hash for c in advisory_commit]
         assert commit.vcs_url in [c.vcs_url for c in advisory_commit]
+
+
+class TestLatestAdvisoryV2(TestCase):
+    def setUp(self):
+        self.logger = TestLogger()
+        self.advisory1 = AdvisoryDataV2(
+            summary="Test advisory old",
+            aliases=["CVE-2025-0001"],
+            references=[],
+            severities=[],
+            weaknesses=[],
+            affected_packages=[
+                AffectedPackageV2(
+                    package=PackageURL.from_string("pkg:npm/foobar"),
+                    affected_version_range=VersionRange.from_string("vers:npm/>3.2.1|<4.0.0"),
+                    fixed_version_range=VersionRange.from_string("vers:npm/4.0.0"),
+                    introduced_by_commit_patches=[],
+                    fixed_by_commit_patches=[],
+                ),
+            ],
+            patches=[],
+            advisory_id="GHSA-1234",
+            url="https://example.com/advisory",
+        )
+
+        self.advisory2 = AdvisoryDataV2(
+            summary="Test advisory new",
+            aliases=["CVE-2025-0001"],
+            references=[],
+            severities=[],
+            weaknesses=[],
+            affected_packages=[
+                AffectedPackageV2(
+                    package=PackageURL.from_string("pkg:npm/foobar"),
+                    affected_version_range=VersionRange.from_string("vers:npm/>3.2.1|<4.0.0"),
+                    fixed_version_range=VersionRange.from_string("vers:npm/4.0.0"),
+                    introduced_by_commit_patches=[],
+                    fixed_by_commit_patches=[],
+                ),
+                AffectedPackageV2(
+                    package=PackageURL.from_string("pkg:npm/foobar"),
+                    affected_version_range=None,
+                    fixed_version_range=None,
+                    introduced_by_commit_patches=[],
+                    fixed_by_commit_patches=[
+                        PackageCommitPatchData(
+                            vcs_url="https://foobar.vcs/",
+                            commit_hash="982f801f",
+                        ),
+                    ],
+                ),
+            ],
+            patches=[],
+            advisory_id="GHSA-1234",
+            url="https://example.com/advisory",
+        )
+
+        insert_advisory_v2(
+            advisory=self.advisory1,
+            pipeline_id="test_pipeline_v2",
+            logger=self.logger.write,
+        )
+
+    def test_latest_advisory_update_on_advisory_insert(self):
+        adv_old = AdvisoryV2.objects.get(avid="test_pipeline_v2/GHSA-1234", is_latest=True)
+        insert_advisory_v2(
+            advisory=self.advisory2,
+            pipeline_id="test_pipeline_v2",
+            logger=self.logger.write,
+        )
+        adv_new = AdvisoryV2.objects.get(avid="test_pipeline_v2/GHSA-1234", is_latest=True)
+        self.assertEqual("Test advisory old", adv_old.summary)
+        self.assertEqual("Test advisory new", adv_new.summary)

vulnerabilities/tests/test_api_v3.py

@@ -14,22 +14,26 @@
 from rest_framework.test import APITestCase
 from univers.version_range import PypiVersionRange
 
+from vulnerabilities.importer import AdvisoryDataV2
 from vulnerabilities.models import AdvisoryV2
 from vulnerabilities.models import PackageV2
 from vulnerabilities.pipes.advisory import insert_advisory_v2
+from vulnerabilities.tests.pipelines import TestLogger
 
 
 class APIV3TestCase(APITestCase):
     def setUp(self):
         from vulnerabilities.models import ImpactedPackage
 
-        self.advisory = AdvisoryV2.objects.create(
-            datasource_id="ghsa",
-            advisory_id="GHSA-1234",
-            avid="ghsa/GHSA-1234",
-            unique_content_id="f" * 64,
-            url="https://example.com/advisory",
-            date_collected="2025-07-01T00:00:00Z",
+        self.logger = TestLogger()
+        self.advisory = insert_advisory_v2(
+            advisory=AdvisoryDataV2(
+                summary="summary",
+                advisory_id="GHSA-1234",
+                url="https://example.com/advisory",
+            ),
+            pipeline_id="ghsa",
+            logger=self.logger.write,
         )
 
         self.package = PackageV2.objects.from_purl(purl="pkg:pypi/sample@1.0.0")