feat: VCIO v3 advisories migration

component_catalog/api.py

@@ -630,7 +630,7 @@ class PackageSerializer(
         read_only=True,
         many=True,
         fields=[
-            "vulnerability_id",
+            "advisory_uid",
             "api_url",
             "uuid",
         ],

component_catalog/templates/component_catalog/tabs/tab_vulnerabilities.html

@@ -51,11 +51,11 @@
           <strong>
             {% if vulnerability.resource_url %}
               <a href="{{ vulnerability.resource_url }}" target="_blank">
-                {{ vulnerability.vulnerability_id }}
+                {{ vulnerability.advisory_id }}
                 <i class="fa-solid fa-up-right-from-square mini"></i>
               </a>
             {% else %}
-              {{ vulnerability.vulnerability_id }}
+              {{ vulnerability.advisory_id }}
             {% endif %}
           </strong>
           <div class="mt-2">

component_catalog/tests/test_api.py

@@ -1351,7 +1351,7 @@ def test_api_package_endpoint_vulnerabilities_features(self):
         self.assertEqual("9.0", results[0]["risk_score"])
         self.assertEqual(
             vulnerability1.vulnerability_id,
-            results[0]["affected_by_vulnerabilities"][0]["vulnerability_id"],
+            results[0]["affected_by_vulnerabilities"][0]["advisory_uid"],
         )
 
         data = {"affected_by": vulnerability1.vulnerability_id}

component_catalog/tests/test_views.py

@@ -1054,7 +1054,7 @@ def test_component_details_view_tab_vulnerabilities(self):
         )
         self.assertContains(response, expected)
         self.assertContains(response, 'id="tab_vulnerabilities"')
-        self.assertContains(response, vulnerability1.vcid)
+        self.assertContains(response, vulnerability1.vulnerability_id)
 
     def test_component_catalog_component_create_ajax_view(self):
         component_create_ajax_url = reverse("component_catalog:component_add_ajax")
@@ -3020,7 +3020,7 @@ def test_package_details_view_tab_vulnerabilities(self):
         )
         self.assertContains(response, expected)
         self.assertContains(response, 'id="tab_vulnerabilities"')
-        self.assertContains(response, self.vulnerability1.vcid)
+        self.assertContains(response, self.vulnerability1.vulnerability_id)
 
     def test_vulnerablecode_get_plain_purls(self):
         purls = get_plain_purls(packages=[])
@@ -3064,37 +3064,6 @@ def test_vulnerablecode_get_vulnerable_purls(self):
             vulnerable_purls = vulnerablecode.get_vulnerable_purls(packages=[self.package1])
             self.assertEqual(["pkg:pypi/django@2.1"], vulnerable_purls)
 
-    def test_vulnerablecode_get_vulnerable_cpes(self):
-        vulnerablecode = VulnerableCode(self.dataspace)
-        vulnerable_cpes = vulnerablecode.get_vulnerable_cpes(components=[])
-        self.assertEqual([], vulnerable_cpes)
-
-        components = [self.component1, self.component2]
-        vulnerable_cpes = vulnerablecode.get_vulnerable_cpes(components=components)
-        self.assertEqual([], vulnerable_cpes)
-
-        self.component1.cpe = "cpe:2.3:a:djangoproject:django:0.95:*:*:*:*:*:*:*"
-        self.component1.save()
-
-        with mock.patch(
-            "dejacode_toolkit.vulnerablecode.VulnerableCode.bulk_search_by_cpes"
-        ) as bulk_search:
-            bulk_search.return_value = [
-                {
-                    "vulnerability_id": "VCID-188m-1bke-aaae",
-                    "summary": "The administrative interface in django.contrib.admin ",
-                    "references": [
-                        {"reference_id": ""},
-                    ],
-                }
-            ]
-            vulnerable_cpes = vulnerablecode.get_vulnerable_cpes(components=components)
-            self.assertEqual([], vulnerable_cpes)
-
-            bulk_search.return_value[0]["references"] = [{"reference_id": self.component1.cpe}]
-            vulnerable_cpes = vulnerablecode.get_vulnerable_cpes(components=components)
-            self.assertEqual([self.component1.cpe], vulnerable_cpes)
-
     @mock.patch("dejacode_toolkit.vulnerablecode.VulnerableCode.request_get")
     def test_vulnerablecode_get_vulnerabilities_cache(self, mock_request_get):
         vulnerablecode = VulnerableCode(self.dataspace)

dejacode_toolkit/__init__.py

@@ -27,7 +27,7 @@ def get_settings(var_name, default=None):
 def is_service_available(label, session, url, raise_exceptions):
     """Check if a configured integration service is available."""
     try:
-        response = session.head(url, timeout=REQUESTS_TIMEOUT)
+        response = session.head(url, allow_redirects=True, timeout=REQUESTS_TIMEOUT)
         response.raise_for_status()
     except requests.exceptions.RequestException as request_exception:
         logger.debug(f"{label} is_available() error: {request_exception}")
@@ -43,6 +43,7 @@ class BaseService:
     settings_prefix = None
     url_field_name = None
     api_key_field_name = None
+    api_version = None
     default_timeout = REQUESTS_TIMEOUT
 
     def __init__(self, dataspace):
@@ -71,6 +72,9 @@ def __init__(self, dataspace):
 
         self.api_url = f"{self.service_url.rstrip('/')}/api/"
 
+        if self.api_version:
+            self.api_url = f"{self.api_url}{self.api_version.rstrip('/')}/"
+
     def get_session(self):
         session = requests.Session()
 

dejacode_toolkit/vulnerablecode.py

@@ -19,87 +19,44 @@ class VulnerableCode(BaseService):
     settings_prefix = "VULNERABLECODE"
     url_field_name = "vulnerablecode_url"
     api_key_field_name = "vulnerablecode_api_key"
+    api_version = "v3"
 
-    def get_vulnerabilities(
+    def get_vulnerabilities_by_purl(
         self,
-        url,
-        field_name,
-        field_value,
+        purl,
         timeout=None,
     ):
-        """Get list of vulnerabilities."""
-        cached_results = cache.get(field_value)
+        """Get list of vulnerabilities providing a package `purl`."""
+        plain_purl = get_plain_purl(purl)
+
+        cached_results = cache.get(plain_purl)
         if cached_results:
             return cached_results
 
-        payload = {field_name: field_value}
-
-        response = self.request_get(url=url, params=payload, timeout=timeout)
+        response = self.bulk_search_by_purl(purls=[plain_purl], timeout=timeout)
         if response and response.get("count"):
             results = response["results"]
-            cache.set(field_value, results)
+            cache.set(plain_purl, results)
             return results
 
-    def get_vulnerabilities_by_purl(
-        self,
-        purl,
-        timeout=None,
-    ):
-        """Get list of vulnerabilities providing a package `purl`."""
-        return self.get_vulnerabilities(
-            url=f"{self.api_url}packages/",
-            field_name="purl",
-            field_value=get_plain_purl(purl),
-            timeout=timeout,
-        )
-
-    def get_vulnerabilities_by_cpe(
-        self,
-        cpe,
-        timeout=None,
-    ):
-        """Get list of vulnerabilities providing a package or component `cpe`."""
-        return self.get_vulnerabilities(
-            url=f"{self.api_url}cpes/",
-            field_name="cpe",
-            field_value=cpe,
-            timeout=timeout,
-        )
-
     def bulk_search_by_purl(
         self,
         purls,
-        purl_only,
+        details=True,
         timeout=None,
     ):
         """Bulk search of vulnerabilities using the provided list of `purls`."""
-        url = f"{self.api_url}packages/bulk_search"
+        url = f"{self.api_url}packages"
 
         data = {
             "purls": purls,
-            "purl_only": purl_only,
-            "plain_purl": True,
+            "details": details,
         }
 
         logger.debug(f"VulnerableCode: url={url} purls_count={len(purls)}")
         return self.request_post(url=url, json=data, timeout=timeout)
 
-    def bulk_search_by_cpes(
-        self,
-        cpes,
-        timeout=None,
-    ):
-        """Bulk search of vulnerabilities using the provided list of `cpes`."""
-        url = f"{self.api_url}cpes/bulk_search"
-
-        data = {
-            "cpes": cpes,
-        }
-
-        logger.debug(f"VulnerableCode: url={url} cpes_count={len(cpes)}")
-        return self.request_post(url, json=data, timeout=timeout)
-
-    def get_vulnerable_purls(self, packages, purl_only=True, timeout=10):
+    def get_vulnerable_purls(self, packages, details=False, timeout=10):
         """
         Return a list of PURLs for which at least one `affected_by_vulnerabilities`
         was found in the VulnerableCodeDB for the given list of `packages`.
@@ -110,34 +67,11 @@ def get_vulnerable_purls(self, packages, purl_only=True, timeout=10):
             return []
 
         vulnerable_purls = self.bulk_search_by_purl(
-            plain_purls,
-            purl_only=purl_only,
+            purls=plain_purls,
+            details=details,
             timeout=timeout,
         )
-        return vulnerable_purls or []
-
-    def get_vulnerable_cpes(self, components):
-        """
-        Return a list of vulnerable CPEs found in the VulnerableCodeDB for the given
-        list of `components`.
-        """
-        cpes = [component.cpe for component in components if component.cpe]
-
-        if not cpes:
-            return []
-
-        search_results = self.bulk_search_by_cpes(cpes)
-        if not search_results:
-            return []
-
-        vulnerable_cpes = [
-            reference.get("reference_id")
-            for entry in search_results
-            for reference in entry.get("references")
-            if reference.get("reference_id").startswith("cpe")
-        ]
-
-        return list(set(vulnerable_cpes))
+        return vulnerable_purls.get("results") or []
 
     def get_package_url_available_types(self):
         # Replace by fetching the endpoint once available.

product_portfolio/templates/product_portfolio/compliance/compliance_watchlist_card.html

@@ -44,7 +44,7 @@ <h2 class="card-header fw-bold px-2 h6 d-flex justify-content-between align-item
               <a href="{{ product_url }}?vulnerabilities-weighted_risk_score=low#vulnerabilities" class="badge bg-info-subtle text-info-emphasis{% if product.critical_count or product.high_count or product.medium_count %} ms-1{% endif %}">{{ product.low_count }} low</a>
             {% endif %}
             {% if not product.critical_count and not product.high_count and not product.medium_count and not product.low_count %}
-              <span class="text-body-tertiary">None</span>
+              <span class="text-body-tertiary small">None</span>
             {% endif %}
           </div>
         </div>

product_portfolio/tests/test_api.py

@@ -1157,7 +1157,7 @@ def test_api_productpackage_endpoint_vulnerabilities_features(self):
 
         response = self.client.get(self.pp1_detail_url)
         response_analysis = response.data["vulnerability_analyses"][0]
-        self.assertEqual(vulnerability1.vulnerability_id, response_analysis["vulnerability_id"])
+        self.assertEqual(vulnerability1.advisory_uid, response_analysis["advisory_uid"])
         self.assertEqual(analysis1.state, response_analysis["state"])
         self.assertEqual(analysis1.justification, response_analysis["justification"])
 
@@ -1189,7 +1189,7 @@ def test_api_product_endpoint_vulnerabilities_features(self):
 
         response = self.client.get(self.product1_detail_url)
         response_analysis = response.data["vulnerability_analyses"][0]
-        self.assertEqual(vulnerability1.vulnerability_id, response_analysis["vulnerability_id"])
+        self.assertEqual(vulnerability1.advisory_uid, response_analysis["advisory_uid"])
         self.assertEqual(analysis1.state, response_analysis["state"])
         self.assertEqual(analysis1.justification, response_analysis["justification"])
 

product_portfolio/tests/test_views.py

@@ -328,9 +328,9 @@ def test_product_portfolio_tab_vulnerability_view_packages_row_rendering(self):
         expected = f"""
         <span data-bs-toggle="modal"
               data-bs-target="#vulnerability-analysis-modal"
-              data-vulnerability-id="{vulnerability1.vcid}"
+              data-vulnerability-id="{vulnerability1.vulnerability_id}"
               data-package-identifier="{p1}"
-              data-edit-url="/products/vulnerability_analysis/{pp1.uuid}/{vulnerability1.vcid}/"
+              data-edit-url="/products/vulnerability_analysis/{pp1.uuid}/{vulnerability1.vulnerability_id}/"
         >
         <button type="button" data-bs-toggle="tooltip" title="Edit" class="btn btn-link p-0"
                 aria-label="Edit">
@@ -375,15 +375,15 @@ def test_product_portfolio_tab_vulnerability_risk_threshold(self):
 
         url = product1.get_url("tab_vulnerabilities")
         response = self.client.get(url)
-        self.assertContains(response, vulnerability1.vcid)
-        self.assertContains(response, vulnerability2.vcid)
+        self.assertContains(response, vulnerability1.vulnerability_id)
+        self.assertContains(response, vulnerability2.vulnerability_id)
         self.assertContains(response, "2 results")
         self.assertNotContains(response, "A risk threshold filter at")
 
         product1.update(vulnerabilities_risk_threshold=3.0)
         response = self.client.get(url)
-        self.assertNotContains(response, vulnerability1.vcid)
-        self.assertContains(response, vulnerability2.vcid)
+        self.assertNotContains(response, vulnerability1.vulnerability_id)
+        self.assertContains(response, vulnerability2.vulnerability_id)
         self.assertContains(response, "1 results")
         self.assertContains(response, 'A risk threshold filter at "3.0" is currently applied.')
 
@@ -4153,7 +4153,7 @@ def test_product_portfolio_product_security_compliance_export_view_json(self):
 
         data = json.loads(response.content)
         self.assertEqual(1, len(data))
-        self.assertEqual(vulnerability.vulnerability_id, data[0]["vulnerability_id"])
+        self.assertEqual(vulnerability.advisory_uid, data[0]["advisory_uid"])
         # Aliases stay a real list in JSON, not a comma-joined string.
         self.assertEqual(
             ["CVE-2024-42005", "GHSA-pv4p-cwwg-4rph", "PYSEC-2024-70"],
@@ -4215,8 +4215,8 @@ def test_product_portfolio_product_security_compliance_export_view_yaml(self):
         self.assertIn(".yaml", response["Content-Disposition"])
 
         content = response.content.decode()
-        self.assertIn(vulnerability.vulnerability_id, content)
-        self.assertIn("vulnerability_id", content)
+        self.assertIn(vulnerability.advisory_uid, content)
+        self.assertIn("advisory_uid", content)
 
     def test_product_portfolio_product_security_compliance_export_view_respects_permissions(self):
         self.client.login(username=self.basic_user.username, password="secret")
@@ -4239,5 +4239,5 @@ def test_product_portfolio_product_security_compliance_export_view_respects_perm
         response = self.client.get(url + "?export=json")
         self.assertEqual(200, response.status_code)
         data = json.loads(response.content)
-        vulnerability_ids = [entry["vulnerability_id"] for entry in data]
-        self.assertIn(vulnerability.vulnerability_id, vulnerability_ids)
+        vulnerability_ids = [entry["advisory_uid"] for entry in data]
+        self.assertIn(vulnerability.advisory_uid, vulnerability_ids)

product_portfolio/views.py

@@ -1208,7 +1208,7 @@ class ProductTabVulnerabilitiesView(
         Header("affected_packages", _("Package"), help_text="Affected product packages"),
         Header("weighted_risk_score", _("Risk"), filter="weighted_risk_score"),
         Header(
-            "vulnerability_id",
+            "advisory_uid",
             _("Vulnerabilities"),
             help_text="Vulnerabilities affecting the product package",
         ),
@@ -3180,7 +3180,7 @@ def get_context_data(self, **kwargs):
                 ),
             )
             .filter(product_count__gt=0)
-            .order_by("-risk_score", "-product_count", "vulnerability_id")
+            .order_by("-risk_score", "-product_count", "advisory_id")
         )
 
         context["vulnerabilities_qs"] = vulnerabilities[: self.limit]
@@ -3232,7 +3232,7 @@ class ProductSecurityComplianceExportView(
 
     export_filename = "security_compliance"
     export_fields = {
-        "vulnerability_id": "Vulnerability ID",
+        "advisory_uid": "Vulnerability ID",
         "aliases": "Aliases",
         "summary": "Summary",
         "risk_level": "Risk level",