Add endpoint for purl lookup

vulnerabilities/api.py

@@ -347,6 +347,42 @@ def all(self, request):
         vulnerable_purls = [str(package.package_url) for package in vulnerable_packages]
         return Response(vulnerable_purls)
 
+    @action(detail=False, methods=["post"])
+    def lookup(self, request):
+        """
+        Return the response for exact PackageURL requested for.
+        """
+        purl = request.data.get("purl")
+        if not purl:
+            return Response(
+                status=400,
+                data={"Error": "A 'purl' is required."},
+            )
+        return Response(
+            PackageSerializer(
+                Package.objects.for_purls([purl]), many=True, context={"request": request}
+            ).data
+        )
+
+    @action(detail=False, methods=["post"])
+    def bulk_lookup(self, request):
+        """
+        Return the response for exact PackageURLs requested for.
+        """
+        purls = request.data.get("purls") or []
+        if not purls:
+            return Response(
+                status=400,
+                data={"Error": "A non-empty 'purls' list of PURLs is required."},
+            )
+        return Response(
+            PackageSerializer(
+                Package.objects.for_purls(purls),
+                many=True,
+                context={"request": request},
+            ).data
+        )
+
 
 class VulnerabilityFilterSet(filters.FilterSet):
     class Meta:

vulnerabilities/models.py

@@ -546,6 +546,9 @@ def for_cve(self, cve):
         """
         return self.filter(vulnerabilities__vulnerabilityreference__reference_id__exact=cve)
 
+    def for_purls(self, purls=[]):
+        return Package.objects.filter(package_url__in=purls).distinct()
+
 
 def get_purl_query_lookups(purl):
     """

vulnerabilities/tests/test_api.py

@@ -749,3 +749,98 @@ class TesBanUserAgent(TestCase):
     def test_ban_request_with_bytedance_user_agent(self):
         response = self.client.get(f"/api/packages", format="json", HTTP_USER_AGENT="bytedance")
         assert 404 == response.status_code
+
+
+class TestLookup(TestCase):
+    def setUp(self):
+        Package.objects.create(
+            type="pypi", namespace="", name="microweber/microweber", version="1.2"
+        )
+        self.user = ApiUser.objects.create_api_user(username="e@mail.com")
+        self.auth = f"Token {self.user.auth_token.key}"
+        self.csrf_client = APIClient(enforce_csrf_checks=True)
+        self.csrf_client.credentials(HTTP_AUTHORIZATION=self.auth)
+
+    def test_lookup_endpoint_failure(self):
+        request_body = {"purl": None}
+        response = self.csrf_client.post(
+            "/api/packages/lookup",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+        assert response == {"Error": "A 'purl' is required."}
+
+    def test_lookup_endpoint(self):
+        request_body = {"purl": "pkg:pypi/microweber/microweber@1.2"}
+        response = self.csrf_client.post(
+            "/api/packages/lookup",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+        assert len(response) == 1
+        assert response[0]["purl"] == "pkg:pypi/microweber/microweber@1.2"
+
+    def test_lookup_endpoint_1(self):
+        Package.objects.create(
+            type="pypi", namespace="microweber", name="microweber", version="1.2"
+        )
+        request_body = {"purl": "pkg:pypi/microweber/microweber@1.2"}
+        response = self.csrf_client.post(
+            "/api/packages/lookup",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+        assert len(response) == 2
+        assert response[0]["purl"] == "pkg:pypi/microweber/microweber@1.2"
+        assert response[0]["purl"] == response[1]["purl"]
+
+    def test_lookup_endpoint_with_one_purl_with_qualifier(self):
+        Package.objects.create(
+            type="pypi", namespace="microweber", name="microweber", version="1.2"
+        )
+        Package.objects.create(
+            type="pypi",
+            namespace="microweber",
+            name="microweber",
+            version="1.2",
+            qualifiers={"foo": "bar"},
+        )
+        request_body = {"purl": "pkg:pypi/microweber/microweber@1.2"}
+        response = self.csrf_client.post(
+            "/api/packages/lookup",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+        assert len(response) == 2
+        assert response[0]["purl"] == "pkg:pypi/microweber/microweber@1.2"
+        assert response[0]["purl"] == response[1]["purl"]
+        request_body = {"purl": "pkg:pypi/microweber/microweber@1.2?foo=bar"}
+        response = self.csrf_client.post(
+            "/api/packages/lookup",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+        assert len(response) == 1
+        assert response[0]["purl"] == "pkg:pypi/microweber/microweber@1.2?foo=bar"
+        request_body = {"purl": "pkg:pypi/microweber/microweber@1.2?foo=baz"}
+        response = self.csrf_client.post(
+            "/api/packages/lookup",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+        assert response == []
+
+    def test_bulk_lookup_endpoint(self):
+        request_body = {
+            "purls": [
+                "pkg:pypi/microweber/microweber@1.2?foo=bar",
+                "pkg:pypi/microweber/microweber@1.2",
+                "pkg:pypi/foo/bar@1.0",
+            ],
+        }
+        response = self.csrf_client.post(
+            "/api/packages/bulk_lookup",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+        assert len(response) == 1