Add vulnerabilities REST API endpoint #104

CHANGELOG.rst

@@ -29,6 +29,15 @@ Release notes
   analysed are listed and can be selected for "analysis propagation".
   https://github.com/aboutcode-org/dejacode/issues/105
 
+- Add vulnerabilities REST API endpoint that mimics the content and features of the
+  vulnerabilities list view.
+  Add `risk_score` and `affected_by_vulnerabilities` fields in Package endpoint.
+  Add `vulnerability_analyses` field in Product and ProductPackage endpoints.
+  Add `is_vulnerable` and `affected_by` filters in Product, Package, and ProductPackage
+  endpoints.
+  Add `risk_score` filter in Package endpoint.
+  https://github.com/aboutcode-org/dejacode/issues/104
+
 ### Version 5.2.1
 
 - Fix the models documentation navigation.

component_catalog/api.py

@@ -24,6 +24,7 @@
 
 from component_catalog.admin import ComponentAdmin
 from component_catalog.admin import PackageAdmin
+from component_catalog.filters import IsVulnerableFilter
 from component_catalog.fuzzy import FuzzyPackageNameSearch
 from component_catalog.license_expression_dje import get_license_objects
 from component_catalog.license_expression_dje import normalize_and_validate_expression
@@ -54,6 +55,9 @@
 from dje.views import SendAboutFilesMixin
 from license_library.models import License
 from organization.api import OwnerEmbeddedSerializer
+from vulnerabilities.api import VulnerabilitySerializer
+from vulnerabilities.filters import RISK_SCORE_RANGES
+from vulnerabilities.filters import ScoreRangeFilter
 
 
 class LicenseSummaryMixin:
@@ -426,6 +430,13 @@ class ComponentFilterSet(DataspacedAPIFilterSet):
     name_version = NameVersionFilter(
         label="Name:Version",
     )
+    is_vulnerable = IsVulnerableFilter(
+        field_name="affected_by_vulnerabilities",
+    )
+    affected_by = django_filters.CharFilter(
+        field_name="affected_by_vulnerabilities__vulnerability_id",
+        label="Affected by (vulnerability_id)",
+    )
 
     class Meta:
         model = Component
@@ -450,6 +461,8 @@ class Meta:
             "last_modified_date",
             "name_version",
             "keywords",
+            "is_vulnerable",
+            "affected_by",
         )
 
 
@@ -610,6 +623,15 @@ class PackageSerializer(
         required=False,
         allow_null=True,
     )
+    affected_by_vulnerabilities = VulnerabilitySerializer(
+        read_only=True,
+        many=True,
+        fields=[
+            "vulnerability_id",
+            "api_url",
+            "uuid",
+        ],
+    )
 
     class Meta:
         model = Package
@@ -669,6 +691,8 @@ class Meta:
             "created_date",
             "last_modified_date",
             "collect_data",
+            "risk_score",
+            "affected_by_vulnerabilities",
         )
         extra_kwargs = {
             "api_url": {
@@ -777,6 +801,14 @@ class PackageAPIFilterSet(DataspacedAPIFilterSet):
     last_modified_date = LastModifiedDateFilter()
     fuzzy = FuzzyPackageNameSearch(widget=HiddenInput)
     purl = PackageURLFilter(label="Package URL")
+    is_vulnerable = IsVulnerableFilter(
+        field_name="affected_by_vulnerabilities",
+    )
+    affected_by = django_filters.CharFilter(
+        field_name="affected_by_vulnerabilities__vulnerability_id",
+        label="Affected by (vulnerability_id)",
+    )
+    risk_score = ScoreRangeFilter(score_ranges=RISK_SCORE_RANGES)
 
     class Meta:
         model = Package
@@ -801,6 +833,9 @@ class Meta:
             "last_modified_date",
             "fuzzy",
             "purl",
+            "is_vulnerable",
+            "affected_by",
+            "risk_score",
         )
 
 
@@ -877,6 +912,7 @@ def get_queryset(self):
             .prefetch_related(
                 "component_set__owner",
                 "licenses__category",
+                "affected_by_vulnerabilities",
                 external_references_prefetch,
             )
         )

component_catalog/tests/test_api.py

@@ -45,6 +45,7 @@
 from license_library.models import LicenseChoice
 from organization.models import Owner
 from policy.models import UsagePolicy
+from vulnerabilities.tests import make_vulnerability
 
 
 @override_settings(
@@ -1034,6 +1035,14 @@ def test_api_package_list_endpoint_filters(self):
         self.assertContains(response, self.package1_detail_url)
         self.assertNotContains(response, self.package2_detail_url)
 
+        self.package1.risk_score = 9.0
+        self.package1.save()
+        data = {"risk_score": "critical"}
+        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)
+
     def test_api_package_list_endpoint_multiple_char_filters(self):
         self.client.login(username="super_user", password="secret")
         filters = "?md5={}&md5={}".format(self.package1.md5, self.package2.md5)
@@ -1325,6 +1334,37 @@ def test_api_package_endpoint_update_put(self):
         self.assertEqual(self.base_user, self.package1.created_by)
         self.assertEqual(self.super_user, self.package1.last_modified_by)
 
+    def test_api_package_endpoint_vulnerabilities_features(self):
+        self.client.login(username="super_user", password="secret")
+        vulnerability1 = make_vulnerability(self.dataspace, affecting=self.package1)
+        vulnerability2 = make_vulnerability(self.dataspace)
+        self.package1.update(risk_score=9.0)
+
+        data = {"is_vulnerable": "yes"}
+        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)
+
+        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"],
+        )
+
+        data = {"affected_by": vulnerability1.vulnerability_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}
+        response = self.client.get(self.package_list_url, data)
+        self.assertEqual(0, response.data["count"])
+        self.assertNotContains(response, self.package1_detail_url)
+        self.assertNotContains(response, self.package2_detail_url)
+
     def test_api_package_license_choices_fields(self):
         self.client.login(username="super_user", password="secret")
 

dejacode/urls.py

@@ -52,6 +52,7 @@
 from product_portfolio.api import ProductPackageViewSet
 from product_portfolio.api import ProductViewSet
 from reporting.api import ReportViewSet
+from vulnerabilities.api import VulnerabilityViewSet
 from workflow.api import RequestTemplateViewSet
 from workflow.api import RequestViewSet
 
@@ -78,6 +79,7 @@
 api_router.register("reports", ReportViewSet)
 api_router.register("external_references", ExternalReferenceViewSet)
 api_router.register("usage_policies", UsagePolicyViewSet)
+api_router.register("vulnerabilities", VulnerabilityViewSet)
 
 
 urlpatterns = [

dje/api.py

@@ -226,7 +226,32 @@ def get_permissions(self):
         return permission_classes + extra_permission
 
 
-class DataspacedSerializer(serializers.HyperlinkedModelSerializer):
+class DynamicFieldsSerializerMixin:
+    """
+    A Serializer mixin that takes an additional `fields` or `exclude_fields`
+    arguments to customize the field selection.
+
+    Inspired by https://www.django-rest-framework.org/api-guide/serializers/#example
+    """
+
+    def __init__(self, *args, **kwargs):
+        fields = kwargs.pop("fields", [])
+        exclude_fields = kwargs.pop("exclude_fields", [])
+
+        super().__init__(*args, **kwargs)
+
+        if fields:
+            self.fields = {
+                field_name: field
+                for field_name, field in self.fields.items()
+                if field_name in fields
+            }
+
+        for field_name in exclude_fields:
+            self.fields.pop(field_name)
+
+
+class DataspacedSerializer(DynamicFieldsSerializerMixin, serializers.HyperlinkedModelSerializer):
     def __init__(self, *args, **kwargs):
         """
         Add the `dataspace` attribute from the request User Dataspace.

dje/tests/test_outputs.py

@@ -20,8 +20,8 @@
 from dje.tests import create_user
 from product_portfolio.models import Product
 from product_portfolio.tests import make_product_package
-from vulnerabilities.models import VulnerabilityAnalysis
 from vulnerabilities.tests import make_vulnerability
+from vulnerabilities.tests import make_vulnerability_analysis
 
 
 class OutputsTestCase(TestCase):
@@ -117,21 +117,19 @@ def test_outputs_get_cyclonedx_bom_include_vex(self):
         self.assertEqual(vulnerability1.vulnerability_id, bom.vulnerabilities[0].id)
         self.assertIsNone(bom.vulnerabilities[0].analysis)
 
-        VulnerabilityAnalysis.objects.create(
-            product_package=product_package1,
-            vulnerability=vulnerability1,
-            state=VulnerabilityAnalysis.State.RESOLVED,
-            justification=VulnerabilityAnalysis.Justification.CODE_NOT_PRESENT,
-            detail="detail",
-            dataspace=self.dataspace,
-        )
+        analysis1 = make_vulnerability_analysis(product_package1, vulnerability1)
         bom = outputs.get_cyclonedx_bom(
             instance=self.product1,
             user=self.super_user,
             include_vex=True,
         )
         analysis = bom.vulnerabilities[0].analysis
-        expected = {"detail": "detail", "justification": "code_not_present", "state": "resolved"}
+        expected = {
+            "detail": analysis1.detail,
+            "justification": str(analysis1.justification),
+            "response": [str(response) for response in analysis1.responses],
+            "state": str(analysis1.state),
+        }
         self.assertEqual(expected, json.loads(analysis.as_json()))
 
     def test_outputs_get_cyclonedx_bom_json(self):