Merge pull request #980 from TG1999/ignore_qualifiers_and_subpath

Ignore qualifiers and subpath from PURL search lookups #978

Author: TG1999

CHANGELOG.rst

@@ -7,6 +7,7 @@ Version v30.2.1
 ----------------
 
 - We refactored and fixed the LaunchPad API code.
+- We now ignore qualifiers and subpath from PURL search lookups. 
 
 
 Version v30.2.0

vulnerabilities/api.py

@@ -22,6 +22,7 @@
 from vulnerabilities.models import Vulnerability
 from vulnerabilities.models import VulnerabilityReference
 from vulnerabilities.models import VulnerabilitySeverity
+from vulnerabilities.models import get_purl_query_lookups
 
 
 class VulnerabilitySeveritySerializer(serializers.ModelSerializer):
@@ -210,8 +211,8 @@ def filter_purl(self, queryset, name, value):
                 detail={"error": f'"{purl}" is not a valid Package URL: {ve}'},
             )
 
-        attrs = {k: v for k, v in purl.to_dict().items() if v}
-        return self.queryset.filter(**attrs)
+        lookups = get_purl_query_lookups(purl)
+        return self.queryset.filter(**lookups)
 
 
 class PackageViewSet(viewsets.ReadOnlyModelViewSet):
@@ -236,12 +237,11 @@ def bulk_search(self, request):
         for purl in request.data["purls"]:
             try:
                 purl_string = purl
-                purl = PackageURL.from_string(purl).to_dict()
+                purl = PackageURL.from_string(purl)
             except ValueError:
                 return Response(status=400, data={"Error": f"Invalid Package URL: {purl}"})
-            purl_data = Package.objects.filter(
-                **{key: value for key, value in purl.items() if value}
-            )
+            lookups = get_purl_query_lookups(purl)
+            purl_data = Package.objects.filter(**lookups)
             purl_response = {}
             if purl_data:
                 purl_response = PackageSerializer(purl_data[0], context={"request": request}).data

vulnerabilities/models.py

@@ -218,6 +218,19 @@ def with_vulnerability_counts(self):
         )
 
 
+def get_purl_query_lookups(purl):
+    """
+    Do not reference all the possible qualifiers and relax the
+    purl matching to only lookup the type, namespace, name and version fields.
+    """
+    lookup_fields = ["type", "namespace", "name", "version"]
+    return {
+        field_name: value
+        for field_name, value in purl.to_dict().items()
+        if value and field_name in lookup_fields
+    }
+
+
 class Package(PackageURLMixin):
     """
     A software package with related vulnerabilities.

vulnerabilities/tests/test_fix_api.py

@@ -281,6 +281,13 @@ def test_api_with_all_vulnerable_packages(self):
                 "pkg:generic/nginx/test@9",
             ]
 
+    def test_api_with_ignorning_qualifiers(self):
+        response = self.csrf_client.get(
+            f"/api/packages/?purl=pkg:generic/nginx/test@9?foo=bar", format="json"
+        ).data
+        assert response["count"] == 1
+        assert response["results"][0]["purl"] == "pkg:generic/nginx/test@9"
+
 
 class CPEApi(TestCase):
     def setUp(self):
@@ -353,7 +360,7 @@ def setUp(self):
             attrs = {k: v for k, v in purl.to_dict().items() if v}
             Package.objects.create(**attrs)
 
-    def test_api_response(self):
+    def test_bulk_api_response(self):
         request_body = {
             "purls": self.packages,
         }
@@ -364,6 +371,30 @@ def test_api_response(self):
         ).json()
         assert len(response) == 13
 
+    def test_bulk_api_response_with_ignoring_qualifiers(self):
+        request_body = {
+            "purls": ["pkg:nginx/[email protected]?qualifiers=dev"],
+        }
+        response = self.csrf_client.post(
+            "/api/packages/bulk_search",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+        assert len(response) == 1
+        assert response[0]["purl"] == "pkg:nginx/[email protected]"
+
+    def test_bulk_api_response_with_ignoring_subpath(self):
+        request_body = {
+            "purls": ["pkg:nginx/[email protected]#dev/subpath"],
+        }
+        response = self.csrf_client.post(
+            "/api/packages/bulk_search",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+        assert len(response) == 1
+        assert response[0]["purl"] == "pkg:nginx/[email protected]"
+
 
 class BulkSearchAPICPE(TestCase):
     def setUp(self):

vulnerabilities/tests/test_view.py

@@ -9,14 +9,37 @@
 
 from django.test import Client
 from django.test import TestCase
+from packageurl import PackageURL
 
 from vulnerabilities.models import Alias
+from vulnerabilities.models import Package
 from vulnerabilities.models import Vulnerability
+from vulnerabilities.views import PackageSearch
 
 
 class PackageSearchTestCase(TestCase):
     def setUp(self):
         self.client = Client()
+        packages = [
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+            "pkg:nginx/[email protected]",
+        ]
+        self.packages = packages
+        for package in packages:
+            purl = PackageURL.from_string(package)
+            attrs = {k: v for k, v in purl.to_dict().items() if v}
+            Package.objects.create(**attrs)
 
     def test_packages_search_view_paginator(self):
         response = self.client.get("/packages/search?type=deb&name=&page=1")
@@ -28,6 +51,12 @@ def test_packages_search_view_paginator(self):
         response = self.client.get("/packages/search?type=&name=&page=")
         self.assertEqual(response.status_code, 200)
 
+    def test_package_view(self):
+        qs = PackageSearch().get_queryset(query="pkg:nginx/[email protected]?foo=bar")
+        pkgs = list(qs)
+        self.assertEqual(len(pkgs), 1)
+        self.assertEqual(pkgs[0].purl, "pkg:nginx/[email protected]")
+
 
 class VulnerabilitySearchTestCase(TestCase):
     def setUp(self):

vulnerabilities/views.py

@@ -93,10 +93,6 @@ def get_queryset(self, query=None):
                 qs = qs.filter(name__iexact=name)
             if version:
                 qs = qs.filter(version__iexact=version)
-            if qualifiers:
-                qs = qs.filter(qualifiers=qualifiers)
-            if subpath:
-                qs = qs.filter(subpath__iexact=subpath)
 
         return qs.annotate(
             vulnerability_count=Count(