New content ID pipeline

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/importer.py

@@ -9,6 +9,7 @@
 
 import dataclasses
 import datetime
+import functools
 import logging
 import os
 import shutil
@@ -46,7 +47,8 @@
 logger = logging.getLogger(__name__)
 
 
-@dataclasses.dataclass(frozen=True)
+@dataclasses.dataclass(eq=True)
+@functools.total_ordering
 class VulnerabilitySeverity:
     # FIXME: this should be named scoring_system, like in the model
     system: ScoringSystem
@@ -55,25 +57,23 @@ class VulnerabilitySeverity:
     published_at: Optional[datetime.datetime] = None
 
     def to_dict(self):
-        published_at_dict = (
-            {"published_at": self.published_at.isoformat()} if self.published_at else {}
-        )
-        return {
+        data = {
             "system": self.system.identifier,
             "value": self.value,
             "scoring_elements": self.scoring_elements,
-            **published_at_dict,
         }
-
-    def __eq__(self, other):
-        if not isinstance(other, VulnerabilitySeverity):
-            return NotImplemented
-        return str(self.to_dict()) == str(other.to_dict())
+        if self.published_at:
+            data["published_at"] = self.published_at.isoformat()
+        return data
 
     def __lt__(self, other):
         if not isinstance(other, VulnerabilitySeverity):
             return NotImplemented
-        return str(self.to_dict()) < str(other.to_dict())
+        return self._cmp_key() < other._cmp_key()
+
+    # TODO: Add cache
+    def _cmp_key(self):
+        return (str(self.system), self.value, self.scoring_elements, self.published_at)
 
     @classmethod
     def from_dict(cls, severity: dict):
@@ -89,7 +89,8 @@ def from_dict(cls, severity: dict):
         )
 
 
-@dataclasses.dataclass(frozen=True)
+@dataclasses.dataclass(eq=True)
+@functools.total_ordering
 class Reference:
     reference_id: str = ""
     reference_type: str = ""
@@ -100,31 +101,22 @@ def __post_init__(self):
         if not self.url:
             raise TypeError("Reference must have a url")
 
-    def normalized(self):
-        severities = sorted(self.severities)
-        return Reference(
-            reference_id=self.reference_id,
-            url=self.url,
-            severities=severities,
-            reference_type=self.reference_type,
-        )
-
-    def __eq__(self, other):
-        if not isinstance(other, Reference):
-            return NotImplemented
-        return str(self.to_dict()) == str(other.to_dict())
-
     def __lt__(self, other):
         if not isinstance(other, Reference):
             return NotImplemented
-        return str(self.to_dict()) < str(other.to_dict())
+        return self._cmp_key() < other._cmp_key()
+
+    # TODO: Add cache
+    def _cmp_key(self):
+        return (self.reference_id, self.reference_type, self.url, tuple(self.severities))
 
     def to_dict(self):
+        """Return a normalized dictionary representation"""
         return {
             "reference_id": self.reference_id,
             "reference_type": self.reference_type,
             "url": self.url,
-            "severities": [severity.to_dict() for severity in self.severities],
+            "severities": [severity.to_dict() for severity in sorted(self.severities)],
         }
 
     @classmethod
@@ -160,7 +152,8 @@ class NoAffectedPackages(Exception):
     """
 
 
-@dataclasses.dataclass(frozen=True)
+@functools.total_ordering
+@dataclasses.dataclass(eq=True)
 class AffectedPackage:
     """
     Relate a Package URL with a range of affected versions and a fixed version.
@@ -190,15 +183,14 @@ def get_fixed_purl(self):
             raise ValueError(f"Affected Package {self.package!r} does not have a fixed version")
         return update_purl_version(purl=self.package, version=str(self.fixed_version))
 
-    def __eq__(self, other):
-        if not isinstance(other, AffectedPackage):
-            return NotImplemented
-        return str(self.to_dict()) == str(other.to_dict())
-
     def __lt__(self, other):
         if not isinstance(other, AffectedPackage):
             return NotImplemented
-        return str(self.to_dict()) < str(other.to_dict())
+        return self._cmp_key() < other._cmp_key()
+
+    # TODO: Add cache
+    def _cmp_key(self):
+        return (str(self.package), str(self.affected_version_range), str(self.fixed_version))
 
     @classmethod
     def merge(
@@ -304,7 +296,6 @@ class AdvisoryData:
     date_published: Optional[datetime.datetime] = None
     weaknesses: List[int] = dataclasses.field(default_factory=list)
     url: Optional[str] = None
-    created_by: Optional[str] = None
 
     def __post_init__(self):
         if self.date_published and not self.date_published.tzinfo:

vulnerabilities/migrations/0089_alter_advisory_unique_content_id.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.16 on 2025-02-06 14:27
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0088_fix_alpine_purl_type"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="advisory",
+            name="unique_content_id",
+            field=models.CharField(blank=True, max_length=64),
+        ),
+    ]

vulnerabilities/models.py

@@ -53,7 +53,8 @@
 from vulnerabilities import utils
 from vulnerabilities.severity_systems import EPSS
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
-from vulnerabilities.utils import compute_content_id, normalize_purl
+from vulnerabilities.utils import compute_content_id
+from vulnerabilities.utils import normalize_purl
 from vulnerabilities.utils import purl_to_dict
 from vulnerablecode import __version__ as VULNERABLECODE_VERSION
 
@@ -1315,7 +1316,7 @@ class Advisory(models.Model):
     """
 
     unique_content_id = models.CharField(
-        max_length=32,
+        max_length=64,
         blank=True,
     )
     aliases = models.JSONField(blank=True, default=list, help_text="A list of alias strings")
@@ -1368,10 +1369,10 @@ def save(self, *args, **kwargs):
             checksum.update(value)
         self.unique_content_id = checksum.hexdigest()
         super().save(*args, **kwargs)
-    
+
     def save(self, *args, **kwargs):
         advisory_data = self.to_advisory_data()
-        self.unique_content_id = compute_content_id(advisory_data, include_metadata=False)
+        self.unique_content_id = compute_content_id(advisory_data, include_metadata=False)[:31]
         super().save(*args, **kwargs)
 
     def to_advisory_data(self) -> "AdvisoryData":
@@ -1389,7 +1390,6 @@ def to_advisory_data(self) -> "AdvisoryData":
             date_published=self.date_published,
             weaknesses=self.weaknesses,
             url=self.url,
-            created_by=self.created_by
         )
 
 

vulnerabilities/pipelines/remove_duplicate_advisories.py

@@ -0,0 +1,79 @@
+#
+# 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/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import logging
+from itertools import groupby
+from operator import attrgetter
+
+from django.db.models import Count
+from django.db.models import Q
+
+from vulnerabilities.models import Advisory
+from vulnerabilities.pipelines import VulnerableCodePipeline
+from vulnerabilities.utils import compute_content_id
+
+
+class RemoveDuplicateAdvisoriesPipeline(VulnerableCodePipeline):
+    """Pipeline to remove duplicate advisories based on their content."""
+
+    pipeline_id = "remove_duplicate_advisories"
+
+    @classmethod
+    def steps(cls):
+        return (cls.remove_duplicates,)
+
+    def remove_duplicates(self):
+        """
+        Find advisories with the same content and keep only the latest one.
+        """
+        # Get all advisories that have duplicates based on content ID
+        duplicate_content_ids = (
+            Advisory.objects.values("unique_content_id")
+            .annotate(count=Count("id"))
+            .filter(count__gt=1)
+            .values_list("unique_content_id", flat=True)
+        )
+
+        self.log(
+            f"Found {len(duplicate_content_ids)} content IDs with duplicates", level=logging.INFO
+        )
+
+        for content_id in duplicate_content_ids:
+            # Get all advisories with this content ID
+            advisories = Advisory.objects.filter(unique_content_id=content_id)
+
+            # Find the latest advisory
+            latest = advisories.latest("date_imported")
+
+            # Delete all except the latest
+            advisories.exclude(id=latest.id).delete()
+
+            if self.log:
+                self.log(
+                    f"Kept advisory {latest.id} and removed "
+                    f"{advisories.count() - 1} duplicates for content ID {content_id}",
+                    level=logging.INFO,
+                )
+
+    def update_content_ids(self):
+        """
+        Update content IDs for all advisories that don't have one.
+        """
+        advisories = Advisory.objects.filter(
+            Q(unique_content_id="") | Q(unique_content_id__isnull=True)
+        )
+
+        self.log(f"Found {advisories.count()} advisories without content ID", level=logging.INFO)
+
+        for advisory in advisories:
+            advisory.unique_content_id = compute_content_id(advisory)
+            advisory.save()
+
+            if self.log:
+                self.log(f"Updated content ID for advisory {advisory.id}", level=logging.DEBUG)

vulnerabilities/severity_systems.py

@@ -42,6 +42,9 @@ def compute(self, scoring_elements: str) -> str:
     def get(self, scoring_elements: str):
         return NotImplementedError
 
+    def __str__(self):
+        return f"{self.identifier}"
+
 
 @dataclasses.dataclass(order=True)
 class Cvssv2ScoringSystem(ScoringSystem):