Add plain_purl option and add tests

Signed-off-by: Tushar Goel <[email protected]>

Author: TG1999

vulnerabilities/api.py

@@ -241,22 +241,50 @@ def bulk_search(self, request):
 
         purls = request.data.get("purls", []) or []
         purl_only = request.data.get("purl_only", False)
+        plain_purl = request.data.get("plain_purl", False)
         if not purls or not isinstance(purls, list):
             return Response(
                 status=400,
                 data={"Error": "A non-empty 'purls' list of PURLs is required."},
             )
 
-        query = Package.objects.filter(package_url__in=purls)
+        if plain_purl:
+            purl_objects = [PackageURL.from_string(purl) for purl in purls]
+            plain_purl_objects = [
+                PackageURL(
+                    type=purl.type,
+                    namespace=purl.namespace,
+                    name=purl.name,
+                    version=purl.version,
+                )
+                for purl in purl_objects
+            ]
+            plain_purls = [str(purl) for purl in plain_purl_objects]
+
+            query = (
+                Package.objects.filter(plain_package_url__in=plain_purls)
+                .order_by("plain_package_url")
+                .distinct("plain_package_url")
+            )
+
+            if not purl_only:
+                return Response(
+                    PackageSerializer(query, many=True, context={"request": request}).data
+                )
+
+            # using order by and distinct because there will be
+            # many fully qualified purl for a single plain purl
+            vulnerable_purls = query.vulnerable().only("plain_package_url")
+            vulnerable_purls = [str(package.plain_package_url) for package in vulnerable_purls]
+            return Response(data=vulnerable_purls)
+
+        query = Package.objects.filter(package_url__in=purls).distinct()
 
         if not purl_only:
-            return Response(
-                PackageSerializer(query.distinct(), many=True, context={"request": request}).data
-            )
+            return Response(PackageSerializer(query, many=True, context={"request": request}).data)
 
-        vulnerable_purls = (
-            query.filter(packagerelatedvulnerability__fix=False).only("package").distinct()
-        )
+        vulnerable_purls = query.vulnerable().only("package_url")
+        vulnerable_purls = [str(package.package_url) for package in vulnerable_purls]
         return Response(data=vulnerable_purls)
 
     @action(detail=False, methods=["get"], throttle_scope="vulnerable_packages")

vulnerabilities/migrations/0034_package_package_url.py renamed to vulnerabilities/migrations/0034_package_package_url_package_plain_package_url.py

@@ -1,4 +1,4 @@
-# Generated by Django 4.0.7 on 2022-11-25 17:28
+# Generated by Django 4.0.7 on 2022-11-28 12:58
 
 from django.db import migrations, models
 
@@ -13,6 +13,11 @@ class Migration(migrations.Migration):
         migrations.AddField(
             model_name='package',
             name='package_url',
-            field=models.CharField(blank=True, db_index=True, help_text='The Package URL for this package.', max_length=255),
+            field=models.CharField(blank=True, db_index=True, help_text='The Package URL for this package.', max_length=1000),
+        ),
+        migrations.AddField(
+            model_name='package',
+            name='plain_package_url',
+            field=models.CharField(blank=True, db_index=True, help_text='The Package URL for this package without qualifiers and subpath.', max_length=1000),
         ),
     ]

vulnerabilities/migrations/0035_add_package_url_to_packages.py

@@ -15,18 +15,25 @@ def save_purls(apps, schema_editor):
                 qualifiers=package.qualifiers,
                 subpath=package.subpath,
             )
+            plain_purl = PackageURL(
+                type=package.type,
+                namespace=package.namespace,
+                name=package.name,
+                version=package.version,
+            ) 
             package.package_url = str(purl)
+            package.plain_package_url = str(plain_purl)
             updatables.append(package)
 
         updated = Package.objects.bulk_update(
             objs = updatables,
-            fields=["package_url"], 
+            fields=["package_url", "plain_package_url"], 
             batch_size=500,
         )
         print(f"Migrated {updated} packages with package_url")            
 
     dependencies = [
-        ('vulnerabilities', '0034_package_package_url'),
+        ("vulnerabilities", "0034_package_package_url_package_plain_package_url"),
     ]
 
     operations = [

vulnerabilities/migrations/0036_alter_package_package_url.py renamed to vulnerabilities/migrations/0036_alter_package_package_url_and_more.py

@@ -1,4 +1,4 @@
-# Generated by Django 4.0.7 on 2022-11-25 17:34
+# Generated by Django 4.0.7 on 2022-11-28 13:00
 
 from django.db import migrations, models
 
@@ -13,6 +13,11 @@ class Migration(migrations.Migration):
         migrations.AlterField(
             model_name='package',
             name='package_url',
-            field=models.CharField(db_index=True, help_text='The Package URL for this package.', max_length=255),
+            field=models.CharField(db_index=True, help_text='The Package URL for this package.', max_length=1000),
+        ),
+        migrations.AlterField(
+            model_name='package',
+            name='plain_package_url',
+            field=models.CharField(db_index=True, help_text='The Package URL for this package without qualifiers and subpath.', max_length=1000),
         ),
     ]

vulnerabilities/models.py

@@ -532,12 +532,19 @@ class Package(PackageURLMixin):
     )
 
     package_url = models.CharField(
-        max_length=255,
+        max_length=1000,
         null=False,
         help_text="The Package URL for this package.",
         db_index=True,
     )
 
+    plain_package_url = models.CharField(
+        max_length=1000,
+        null=False,
+        help_text="The Package URL for this package without qualifiers and subpath.",
+        db_index=True,
+    )
+
     objects = PackageQuerySet.as_manager()
 
     def save(self, *args, **kwargs):
@@ -549,7 +556,14 @@ def save(self, *args, **kwargs):
             qualifiers=self.qualifiers,
             subpath=self.subpath,
         )
+        plain_purl = PackageURL(
+            type=self.type,
+            namespace=self.namespace,
+            name=self.name,
+            version=self.version,
+        )
         self.package_url = str(purl_object)
+        self.plain_package_url = str(plain_purl)
         super().save(*args, **kwargs)
 
     @property

vulnerabilities/tests/test_fix_api.py

@@ -358,6 +358,19 @@ def setUp(self):
             attrs = {k: v for k, v in purl.to_dict().items() if v}
             Package.objects.create(**attrs)
 
+        vulnerable_packages = [
+            "pkg:nginx/[email protected]?foo=bar",
+            "pkg:nginx/[email protected]?foo=baz",
+        ]
+
+        vuln = Vulnerability.objects.create(summary="test")
+
+        for package in vulnerable_packages:
+            purl = PackageURL.from_string(package)
+            attrs = {k: v for k, v in purl.to_dict().items() if v}
+            pkg = Package.objects.create(**attrs)
+            PackageRelatedVulnerability.objects.create(package=pkg, vulnerability=vuln)
+
     def test_bulk_api_response(self):
         request_body = {
             "purls": self.packages,
@@ -370,9 +383,7 @@ def test_bulk_api_response(self):
         assert len(response) == 13
 
     def test_bulk_api_response_with_ignoring_qualifiers(self):
-        request_body = {
-            "purls": ["pkg:nginx/[email protected]?qualifiers=dev"],
-        }
+        request_body = {"purls": ["pkg:nginx/[email protected]?qualifiers=dev"], "plain_purl": True}
         response = self.csrf_client.post(
             "/api/packages/bulk_search",
             data=json.dumps(request_body),
@@ -382,16 +393,28 @@ def test_bulk_api_response_with_ignoring_qualifiers(self):
         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"], "plain_purl": True}
+        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_with_purl_only_option(self):
         request_body = {
             "purls": ["pkg:nginx/[email protected]#dev/subpath"],
+            "purl_only": True,
+            "plain_purl": True,
         }
         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]"
+        assert response[0] == "pkg:nginx/[email protected]"
 
 
 class BulkSearchAPICPE(TestCase):