diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 261b2cfe6..5c0c7f9b1 100644 --- a/CHANGELOG.rst +++ b/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 diff --git a/vulnerabilities/api.py b/vulnerabilities/api.py index 12da1ed6f..5c949cda4 100644 --- a/vulnerabilities/api.py +++ b/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 diff --git a/vulnerabilities/models.py b/vulnerabilities/models.py index 00c1b9084..549d7f7a5 100644 --- a/vulnerabilities/models.py +++ b/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. diff --git a/vulnerabilities/tests/test_fix_api.py b/vulnerabilities/tests/test_fix_api.py index ef825f5d2..4b13fccc9 100644 --- a/vulnerabilities/tests/test_fix_api.py +++ b/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/nginx@1.0.15?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/nginx@1.0.15" + + def test_bulk_api_response_with_ignoring_subpath(self): + request_body = { + "purls": ["pkg:nginx/nginx@1.0.15#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/nginx@1.0.15" + class BulkSearchAPICPE(TestCase): def setUp(self): diff --git a/vulnerabilities/tests/test_view.py b/vulnerabilities/tests/test_view.py index 7b96ecfd5..e0a655367 100644 --- a/vulnerabilities/tests/test_view.py +++ b/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/nginx@0.6.18", + "pkg:nginx/nginx@1.20.0", + "pkg:nginx/nginx@1.21.0", + "pkg:nginx/nginx@1.20.1", + "pkg:nginx/nginx@1.9.5", + "pkg:nginx/nginx@1.17.2", + "pkg:nginx/nginx@1.17.3", + "pkg:nginx/nginx@1.16.1", + "pkg:nginx/nginx@1.15.5", + "pkg:nginx/nginx@1.15.6", + "pkg:nginx/nginx@1.14.1", + "pkg:nginx/nginx@1.0.7", + "pkg:nginx/nginx@1.0.15", + ] + 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/nginx@1.0.15?foo=bar") + pkgs = list(qs) + self.assertEqual(len(pkgs), 1) + self.assertEqual(pkgs[0].purl, "pkg:nginx/nginx@1.0.15") + class VulnerabilitySearchTestCase(TestCase): def setUp(self): diff --git a/vulnerabilities/views.py b/vulnerabilities/views.py index 9ae9fb4a6..c9f99516f 100644 --- a/vulnerabilities/views.py +++ b/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(