Allow querying /api/packages by Package URL

Signed-off-by: Haiko Schol <hs@haikoschol.com>

Author: haikoschol

vulnerabilities/api.py

@@ -22,7 +22,11 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
 from rest_framework import serializers
+from rest_framework import status
 from rest_framework import viewsets
+from rest_framework.response import Response
+
+from packageurl import PackageURL
 
 from vulnerabilities.models import Package
 from vulnerabilities.models import PackageReference
@@ -82,3 +86,26 @@ class PackageViewSet(viewsets.ReadOnlyModelViewSet):
     queryset = Package.objects.all()
     serializer_class = PackageSerializer
     filter_fields = ('name', 'version')
+
+    def filter_queryset(self, qs):
+        purl = self.request.query_params.get('package_url')
+        if not purl:
+            return super().filter_queryset(qs)
+
+        try:
+            purl = PackageURL.from_string(purl)
+        except ValueError as ve:
+            raise serializers.ValidationError(
+                detail={'error': f'"{purl}" is not a valid Package URL: {ve}'},
+            )
+
+        # Remove "qualifiers" here because it is stored as one string in the model.
+        # For example, a row in the database could have the "qualifiers" column
+        # stored as "foo=bar&spam=eggs". If an API request contains a PURL with
+        # "spam=eggs&foo=bar", the DB query would not include that row.
+        attrs = {k: v for k, v in purl.to_dict().items() if v and k != 'qualifiers'}
+
+        # TODO
+        # Since we are filtering on all the Package URL fields except "qualifiers",
+        # we'll eventually need database indices on them.
+        return self.queryset.filter(**attrs)

vulnerabilities/tests/test_api.py

@@ -47,26 +47,64 @@ def setUpTestData(cls):
         extract_data = debian.extract_vulnerabilities(test_data)
         debian_dump(extract_data)
 
-    def setUp(self):
-        self.response = self.client.get('/api/packages/?name=mimetex', format='json').data
+        Package.objects.create(
+            name='mimetex',
+            version='1.50-1.1',
+            type='deb',
+            namespace='ubuntu'
+        )
 
-    def test_count(self):
-        self.assertEqual(4, self.response['count'])
+    def test_query_by_name(self):
+        response = self.client.get('/api/packages/?name=mimetex', format='json').data
 
-    def test_name(self):
-        first_result = self.response['results'][0]
+        self.assertEqual(5, response['count'])
+
+        first_result = response['results'][0]
         self.assertEqual('mimetex', first_result['name'])
 
-    def test_version(self):
-        versions = {r['version'] for r in self.response['results']}
+        versions = {r['version'] for r in response['results']}
         self.assertIn('1.50-1.1', versions)
         self.assertIn('1.74-1', versions)
 
-    def test_packageurl(self):
-        purls = {r['package_url'] for r in self.response['results']}
+        purls = {r['package_url'] for r in response['results']}
         self.assertIn('pkg:deb/debian/mimetex@1.50-1.1?distro=jessie', purls)
         self.assertIn('pkg:deb/debian/mimetex@1.74-1?distro=jessie', purls)
 
+    def test_query_by_invalid_package_url(self):
+        url = '/api/packages/?package_url=invalid_purl'
+        response = self.client.get(url, format='json')
+
+        self.assertEqual(400, response.status_code)
+        self.assertIn('error', response.data)
+        error = response.data['error']
+        self.assertIn('invalid_purl', error)
+
+    def test_query_by_package_url(self):
+        url = '/api/packages/?package_url=pkg:deb/debian/mimetex@1.50-1.1?distro=jessie'
+        response = self.client.get(url, format='json').data
+
+        self.assertEqual(2, response['count'])
+
+        first_result = response['results'][0]
+        self.assertEqual('mimetex', first_result['name'])
+
+        versions = {r['version'] for r in response['results']}
+        self.assertIn('1.50-1.1', versions)
+        self.assertNotIn('1.74-1', versions)
+
+    def test_query_by_package_url_without_namespace(self):
+        url = '/api/packages/?package_url=pkg:deb/mimetex@1.50-1.1'
+        response = self.client.get(url, format='json').data
+
+        self.assertEqual(3, response['count'])
+
+        first_result = response['results'][0]
+        self.assertEqual('mimetex', first_result['name'])
+
+        purls = {r['package_url'] for r in response['results']}
+        self.assertIn('pkg:deb/debian/mimetex@1.50-1.1?distro=jessie', purls)
+        self.assertIn('pkg:deb/ubuntu/mimetex@1.50-1.1', purls)
+
 
 class TestUbuntuResponse(TestCase):
     def test_ubuntu_response(self):