feat: VCIO v3 advisories migration

Makefile

@@ -27,9 +27,11 @@ shell:
 	# Open a bash session in a standalone container (no stack required)
 	docker run -it $(IMAGE_NAME) bash
 
+# make test              - full suite
+# make test k=<pattern>  - filter by name, e.g. make test k=test_name
 test:
-	@echo "-> Run the test suite"
-	${MANAGE} test --noinput --parallel auto
+	${EXEC} web pip install --find-links=thirdparty/dist/ --no-index --no-cache-dir '.[dev]'
+	${MANAGE} test --noinput --parallel auto $(if $(k),-k $(k),)
 
 migrations:
 	@echo "-> Creates new database migrations"

component_catalog/api.py

@@ -436,8 +436,8 @@ class ComponentFilterSet(DataspacedAPIFilterSet):
         field_name="affected_by_vulnerabilities",
     )
     affected_by = django_filters.CharFilter(
-        field_name="affected_by_vulnerabilities__vulnerability_id",
-        label="Affected by (vulnerability_id)",
+        field_name="affected_by_vulnerabilities__advisory_id",
+        label="Affected by (advisory_id)",
     )
 
     class Meta:
@@ -630,7 +630,8 @@ class PackageSerializer(
         read_only=True,
         many=True,
         fields=[
-            "vulnerability_id",
+            "advisory_uid",
+            "advisory_id",
             "api_url",
             "uuid",
         ],
@@ -809,8 +810,8 @@ class PackageAPIFilterSet(DataspacedAPIFilterSet):
         field_name="affected_by_vulnerabilities",
     )
     affected_by = django_filters.CharFilter(
-        field_name="affected_by_vulnerabilities__vulnerability_id",
-        label="Affected by (vulnerability_id)",
+        field_name="affected_by_vulnerabilities__advisory_id",
+        label="Affected by (advisory_id)",
     )
     risk_score = ScoreRangeFilter(score_ranges=RISK_SCORE_RANGES)
 

component_catalog/filters.py

@@ -119,7 +119,7 @@ class ComponentFilterSet(DataspacedFilterSet):
         widget=DropDownRightWidget(link_content='<i class="fas fa-bug"></i>'),
     )
     affected_by = django_filters.CharFilter(
-        field_name="affected_by_vulnerabilities__vulnerability_id",
+        field_name="affected_by_vulnerabilities__advisory_id",
         label=_("Affected by"),
     )
 
@@ -267,7 +267,7 @@ class PackageFilterSet(DataspacedFilterSet):
         widget=DropDownRightWidget(link_content='<i class="fas fa-bug"></i>'),
     )
     affected_by = django_filters.CharFilter(
-        field_name="affected_by_vulnerabilities__vulnerability_id",
+        field_name="affected_by_vulnerabilities__advisory_id",
         label=_("Affected by"),
     )
     affected_by_last_modified_date = django_filters.DateRangeFilter(

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

@@ -1350,17 +1350,17 @@ def test_api_package_endpoint_vulnerabilities_features(self):
         results = response.data["results"]
         self.assertEqual("9.0", results[0]["risk_score"])
         self.assertEqual(
-            vulnerability1.vulnerability_id,
-            results[0]["affected_by_vulnerabilities"][0]["vulnerability_id"],
+            vulnerability1.advisory_id,
+            results[0]["affected_by_vulnerabilities"][0]["advisory_id"],
         )
 
-        data = {"affected_by": vulnerability1.vulnerability_id}
+        data = {"affected_by": vulnerability1.advisory_id}
         response = self.client.get(self.package_list_url, data)
         self.assertEqual(1, response.data["count"])
         self.assertContains(response, self.package1_detail_url)
         self.assertNotContains(response, self.package2_detail_url)
 
-        data = {"affected_by": vulnerability2.vulnerability_id}
+        data = {"affected_by": vulnerability2.advisory_id}
         response = self.client.get(self.package_list_url, data)
         self.assertEqual(0, response.data["count"])
         self.assertNotContains(response, self.package1_detail_url)

component_catalog/tests/test_filters.py

@@ -406,6 +406,6 @@ def test_package_filterset_affected_by_filter(self):
         self.assertIn(package1, filterset.qs)
         self.assertIn(package2, filterset.qs)
 
-        data = {"affected_by": vulnerability1.vulnerability_id}
+        data = {"affected_by": vulnerability1.advisory_id}
         filterset = PackageFilterSet(dataspace=self.dataspace, data=data)
         self.assertQuerySetEqual(filterset.qs, [package1])

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.advisory_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.advisory_id)
 
     def test_vulnerablecode_get_plain_purls(self):
         purls = get_plain_purls(packages=[])
@@ -3056,65 +3056,34 @@ def test_vulnerablecode_get_vulnerable_purls(self):
         with mock.patch(
             "dejacode_toolkit.vulnerablecode.VulnerableCode.bulk_search_by_purl"
         ) as bulk_search:
-            bulk_search.return_value = []
+            bulk_search.return_value = {"count": 0, "results": []}
             vulnerable_purls = vulnerablecode.get_vulnerable_purls(packages=[self.package1])
             self.assertEqual([], vulnerable_purls)
 
-            bulk_search.return_value = ["pkg:pypi/django@2.1"]
+            bulk_search.return_value = {"count": 1, "results": ["pkg:pypi/django@2.1"]}
             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):
+    @mock.patch("dejacode_toolkit.vulnerablecode.VulnerableCode.bulk_search_by_purl")
+    def test_vulnerablecode_get_vulnerabilities_cache(self, mock_bulk_search):
         vulnerablecode = VulnerableCode(self.dataspace)
 
         self.package1.set_package_url("pkg:pypi/django@2.1")
         self.package1.save()
 
-        mock_request_get.return_value = {
+        mock_bulk_search.return_value = {
             "count": 1,
             "results": True,
         }
 
         results = vulnerablecode.get_vulnerabilities_by_purl(self.package1.package_url)
-        self.assertEqual(1, mock_request_get.call_count)
+        self.assertEqual(1, mock_bulk_search.call_count)
         self.assertTrue(results)
 
         results = vulnerablecode.get_vulnerabilities_by_purl(self.package1.package_url)
-        # request.get was only called once since the results are returned from the cached
+        # bulk_search_by_purl was only called once since the results are returned from the cache
         # on the second call of `get_vulnerabilities_by_purl`.
-        self.assertEqual(1, mock_request_get.call_count)
+        self.assertEqual(1, mock_bulk_search.call_count)
         self.assertTrue(results)
 
     def test_send_scan_notification(self):

dejacode/static/css/dejacode_bootstrap.css

@@ -395,7 +395,7 @@ table.packages-table .column-primary_language {
 }
 
 /* -- Vulnerability List -- */
-table.vulnerabilities-table .column-vulnerability_id {
+table.vulnerabilities-table .column-advisory_uid {
   width: 220px;
 }
 table.vulnerabilities-table .column-aliases {
@@ -413,7 +413,7 @@ table.vulnerabilities-table .column-summary {
 }
 
 /* -- Vulnerability tab -- */
-#tab_vulnerabilities .column-vulnerability_id {
+#tab_vulnerabilities .column-advisory_uid {
   width: 230px;
 }
 #tab_vulnerabilities .column-affected_packages {

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()