Migrate VulnerableCode integration to API v3

scancodeio/settings.py

@@ -431,6 +431,9 @@
 VULNERABLECODE_USER = env.str("VULNERABLECODE_USER", default="")
 VULNERABLECODE_PASSWORD = env.str("VULNERABLECODE_PASSWORD", default="")
 VULNERABLECODE_API_KEY = env.str("VULNERABLECODE_API_KEY", default="")
+VULNERABLECODE_USER_AGENT = env.str(
+    "VULNERABLECODE_USER_AGENT", default="VCIO_API_AGENT"
+)
 
 # PurlDB integration
 

scancodeio/static/main.css

@@ -388,12 +388,12 @@ progress.file-upload::before {
 #resource-list th#column-status {
   min-width: 90px;
 }
+th#column-advisory_id {
+  min-width: 220px;
+}
 #message-list th#column-severity {
   min-width: 110px;
 }
-th#column-vulnerability_id {
-  min-width: 220px;
-}
 th#column-summary {
   width: 40%;
 }

scanpipe/forms.py

@@ -512,7 +512,7 @@ class ProjectSettingsForm(forms.ModelForm):
             attrs={
                 "class": "textarea is-dynamic",
                 "rows": 2,
-                "placeholder": "VCID-q4q6-yfng-aaag\nCVE-2024-27351",
+                "placeholder": "ID-q4q6-yfng-aaag\nCVE-2024-27351",
             },
         ),
     )

scanpipe/management/commands/check-compliance.py

@@ -110,8 +110,8 @@ def check_vulnerabilities(self):
         if self.verbosity > 0:
             if vulnerabilities_count:
                 self.stderr.write(f"{vulnerabilities_count} vulnerabilities found:")
-                for vulnerability_id, vulnerability_data in all_vulnerabilities.items():
-                    self.stderr.write(str(vulnerability_id))
+                for advisory_id, vulnerability_data in all_vulnerabilities.items():
+                    self.stderr.write(str(advisory_id))
                     for affected_obj in vulnerability_data.get("affects", []):
                         self.stderr.write(f" > {affected_obj}")
             else:

scanpipe/migrations/0079_apitoken_data.py

@@ -31,9 +31,7 @@ def migrate_api_tokens(apps, schema_editor):
         )
         for user_id, key, created in rows
     ]
-    migrated_tokens = APIToken.objects.bulk_create(tokens_to_create, ignore_conflicts=True)
-    if migrated_tokens:
-        print(f" -> {len(migrated_tokens)} tokens migrated.")
+    APIToken.objects.bulk_create(tokens_to_create, ignore_conflicts=True)
 
 
 def reverse_migrate_api_tokens(apps, schema_editor):

scanpipe/migrations/0080_vulnerablecode_v3_data.py

@@ -0,0 +1,57 @@
+# Generated by Django 6.0.3 on 2026-04-14 11:10
+
+from django.db import migrations
+from django.db.models import Q
+
+
+def add_advisory_id(apps, schema_editor):
+    """
+    Copy vulnerability_id to advisory_uid and advisory_id in
+    affected_by_vulnerabilities entries.
+    """
+    DiscoveredPackage = apps.get_model("scanpipe", "DiscoveredPackage")
+    DiscoveredDependency = apps.get_model("scanpipe", "DiscoveredDependency")
+    EMPTY_VALUES = [None, [], ""]
+
+    vulnerable = ~Q(affected_by_vulnerabilities__in=EMPTY_VALUES)
+
+    for model in [DiscoveredPackage, DiscoveredDependency]:
+        to_update = []
+        for instance in model.objects.filter(vulnerable).only("affected_by_vulnerabilities"):
+            for entry in instance.affected_by_vulnerabilities:
+                entry.setdefault("advisory_uid", entry.get("vulnerability_id", ""))
+                entry.setdefault("advisory_id", entry.get("vulnerability_id", ""))
+            to_update.append(instance)
+        model.objects.bulk_update(to_update, ["affected_by_vulnerabilities"])
+
+
+def remove_advisory_id(apps, schema_editor):
+    """Remove advisory_uid and advisory_id keys from affected_by_vulnerabilities entries."""
+    DiscoveredPackage = apps.get_model("scanpipe", "DiscoveredPackage")
+    DiscoveredDependency = apps.get_model("scanpipe", "DiscoveredDependency")
+    EMPTY_VALUES = [None, [], ""]
+
+    vulnerable = ~Q(affected_by_vulnerabilities__in=EMPTY_VALUES)
+
+    for model in [DiscoveredPackage, DiscoveredDependency]:
+        to_update = []
+        for instance in model.objects.filter(vulnerable).only("affected_by_vulnerabilities"):
+            for entry in instance.affected_by_vulnerabilities:
+                entry.pop("advisory_uid", None)
+                entry.pop("advisory_id", None)
+            to_update.append(instance)
+        model.objects.bulk_update(to_update, ["affected_by_vulnerabilities"])
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('scanpipe', '0079_apitoken_data'),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            add_advisory_id,
+            reverse_code=remove_advisory_id,
+        ),
+    ]

scanpipe/models.py

@@ -1573,7 +1573,7 @@ def vulnerabilities(self):
         """
         Return a dict of all vulnerabilities affecting this project.
 
-        Combines package and dependency vulnerabilities, keyed by vulnerability_id.
+        Combines package and dependency vulnerabilities, keyed by advisory_uid.
         Each vulnerability includes an "affects" list of all affected packages
         and dependencies.
         """
@@ -1583,11 +1583,13 @@ def vulnerabilities(self):
 
         for queryset in querysets:
             vulnerabilities = queryset.get_vulnerabilities_dict()
-            for vcid, vuln_data in vulnerabilities.items():
-                if vcid in vulnerabilities_dict:
-                    vulnerabilities_dict[vcid]["affects"].extend(vuln_data["affects"])
+            for advisory_uid, vuln_data in vulnerabilities.items():
+                if advisory_uid in vulnerabilities_dict:
+                    vulnerabilities_dict[advisory_uid]["affects"].extend(
+                        vuln_data["affects"]
+                    )
                 else:
-                    vulnerabilities_dict[vcid] = vuln_data
+                    vulnerabilities_dict[advisory_uid] = vuln_data
 
         return vulnerabilities_dict
 
@@ -3361,51 +3363,49 @@ def get_vulnerabilities_list(self):
         queryset.
 
         Extracts and flattens the affected_by_vulnerabilities field from
-        all objects in the queryset. Removes duplicates based on vulnerability_id
+        all objects in the queryset. Removes duplicates based on advisory_uid
         while preserving the first occurrence of each unique vulnerability.
         """
         vulnerabilities_lists = self.values_list(self.AFFECTED_BY_FIELD, flat=True)
         flatten_vulnerabilities = chain.from_iterable(vulnerabilities_lists)
 
-        # Deduplicate by vulnerability_id while preserving order
+        # Deduplicate by advisory_uid while preserving order
         unique_vulnerabilities = {
-            vuln["vulnerability_id"]: vuln for vuln in flatten_vulnerabilities
+            vuln["advisory_uid"]: vuln for vuln in flatten_vulnerabilities
         }
 
-        return sorted(
-            unique_vulnerabilities.values(), key=itemgetter("vulnerability_id")
-        )
+        return sorted(unique_vulnerabilities.values(), key=itemgetter("advisory_uid"))
 
     def get_vulnerabilities_dict(self):
         """
-        Return a dict of vulnerabilities keyed by vulnerability_id.
+        Return a dict of vulnerabilities keyed by advisory_uid.
 
         Each vulnerability includes an "affects" list containing all
         objects from this queryset affected by that vulnerability.
 
         Returns:
             dict: {
-                'VCID-1': {
-                    'vulnerability_id': 'VCID-1',
+                'ID-1': {
+                    'advisory_uid': 'ID-1',
                     'affects': [obj1, obj2, ...]
                 },
                 ...
             }
 
         """
-        vulnerabilities_dict = {}
+        vulnerabilities = {}
 
         for obj in self.vulnerable_ordered():
             for vulnerability in obj.affected_by_vulnerabilities:
-                vcid = vulnerability.get("vulnerability_id")
-                if not vcid:
+                advisory_uid = vulnerability.get("advisory_uid")
+                if not advisory_uid:
                     continue
 
-                if vcid not in vulnerabilities_dict:
-                    vulnerabilities_dict[vcid] = {**vulnerability, "affects": []}
-                vulnerabilities_dict[vcid]["affects"].append(obj)
+                if advisory_uid not in vulnerabilities:
+                    vulnerabilities[advisory_uid] = {**vulnerability, "affects": []}
+                vulnerabilities[advisory_uid]["affects"].append(obj)
 
-        return vulnerabilities_dict
+        return vulnerabilities
 
 
 class DiscoveredPackageQuerySet(

scanpipe/pipes/cyclonedx.py

@@ -186,9 +186,11 @@ def cyclonedx_component_to_package_data(
     if affected_by := vulnerabilities.get(bom_ref):
         for cdx_vulnerability in affected_by:
             cdx_vulnerability_json = cdx_vulnerability.as_json(view_=BaseSchemaVersion)
+            cdx_vulnerability_id = str(cdx_vulnerability.id)
             affected_by_vulnerabilities.append(
                 {
-                    "vulnerability_id": str(cdx_vulnerability.id),
+                    "advisory_uid": cdx_vulnerability_id,
+                    "advisory_id": cdx_vulnerability_id,
                     "summary": cdx_vulnerability.description,
                     "cdx_vulnerability_data": json.loads(cdx_vulnerability_json),
                 }

scanpipe/pipes/output.py

@@ -579,7 +579,7 @@ def add_vulnerabilities_sheet(workbook, project):
     ]
 
     vulnerability_fields = [
-        "vulnerability_id",
+        "advisory_id",
         "aliases",
         "summary",
         "risk_score",
@@ -863,7 +863,7 @@ def vulnerability_as_cyclonedx(vulnerability_data, component_bom_ref):
     ]
 
     return cdx_vulnerability.Vulnerability(
-        id=vulnerability_data.get("vulnerability_id"),
+        id=vulnerability_data.get("advisory_id"),
         source=source,
         description=vulnerability_data.get("summary"),
         affects=affects,