Add Live Evaluation API endpoint #1902

* Add a new API endpoint to run live evaluation importers

* Add tests for the live evaluation API endpoint

Signed-off-by: Michael Ehab Mikhail <michael.ehab@hotmail.com>

Author: michaelehab

vulnerabilities/api_v2.py

@@ -8,6 +8,9 @@
 #
 
 
+from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import as_completed
+
 from django.db.models import Prefetch
 from django_filters import rest_framework as filters
 from drf_spectacular.utils import OpenApiParameter
@@ -25,6 +28,7 @@
 from rest_framework.reverse import reverse
 from rest_framework.throttling import AnonRateThrottle
 
+from vulnerabilities.importers import LIVE_IMPORTERS_REGISTRY
 from vulnerabilities.models import AdvisoryReference
 from vulnerabilities.models import AdvisorySeverity
 from vulnerabilities.models import AdvisoryV2
@@ -1225,3 +1229,83 @@ def lookup(self, request):
         return Response(
             AdvisoryPackageV2Serializer(qs, many=True, context={"request": request}).data
         )
+
+
+class LiveEvaluationSerializer(serializers.Serializer):
+    purl_string = serializers.CharField(help_text="PackageURL to evaluate")
+    no_threading = serializers.BooleanField(required=False, default=False)
+
+
+class LiveEvaluationViewSet(viewsets.GenericViewSet):
+    serializer_class = LiveEvaluationSerializer
+
+    @extend_schema(
+        request=LiveEvaluationSerializer,
+        responses={
+            202: {"description": "Live evaluation done successfully"},
+            400: {"description": "Invalid request"},
+            500: {"description": "Internal server error"},
+        },
+    )
+    @action(detail=False, methods=["post"])
+    def evaluate(self, request):
+        serializer = self.get_serializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(
+                serializer.errors,
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        purl_string = serializer.validated_data.get("purl_string")
+        no_threading = serializer.validated_data.get("no_threading", False)
+
+        try:
+            purl = PackageURL.from_string(purl_string) if purl_string else None
+            if not purl:
+                return Response({"error": "Invalid PackageURL"}, status=status.HTTP_400_BAD_REQUEST)
+        except Exception as e:
+            return Response(
+                {"error": f"Invalid PackageURL: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST
+            )
+
+        importers = [
+            importer
+            for importer in LIVE_IMPORTERS_REGISTRY.values()
+            if hasattr(importer, "supported_types")
+            and purl.type in getattr(importer, "supported_types", [])
+        ]
+
+        if not importers:
+            return Response(
+                {"error": f"No live importers found for purl type '{purl.type}'"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        results = []
+
+        def run_importer(importer):
+            importer_name = getattr(importer, "pipeline_id", importer.__name__)
+            response_data = {"importer": importer_name, "purl": purl_string, "steps_completed": []}
+            try:
+                pipeline_instance = importer(purl=purl)
+                status_code, error = pipeline_instance.execute()
+                if status_code != 0:
+                    response_data["error"] = f"Importer {importer_name} failed: {error}"
+                else:
+                    response_data["steps_completed"].append("import")
+            except Exception as e:
+                response_data["error"] = f"Error running importer {importer_name}: {str(e)}"
+            return response_data
+
+        if not no_threading and len(importers) > 1:
+            with ThreadPoolExecutor(max_workers=len(importers)) as executor:
+                future_to_importer = {
+                    executor.submit(run_importer, importer): importer for importer in importers
+                }
+                for future in as_completed(future_to_importer):
+                    results.append(future.result())
+        else:
+            for importer in importers:
+                results.append(run_importer(importer))
+
+        return Response(results, status=status.HTTP_202_ACCEPTED)

vulnerabilities/tests/test_api_v2.py

@@ -905,3 +905,68 @@ def test_get_all_vulnerable_purls(self):
             response = self.client.get(url)
         assert response.status_code == 200
         assert "pkg:pypi/sample@1.0.0" in response.data
+
+
+class LiveEvaluationAPITest(APITestCase):
+    def setUp(self):
+        self.client = APIClient(enforce_csrf_checks=True)
+        self.url = "/api/v2/live-evaluation/evaluate"
+
+    @patch("vulnerabilities.api_v2.LIVE_IMPORTERS_REGISTRY")
+    def test_evaluate_success(self, mock_registry):
+        class MockImporter:
+            pipeline_id = "dummy"
+            supported_types = ["pypi"]
+
+            def __init__(self, purl=None):
+                pass
+
+            def execute(self):
+                return 0, None
+
+        mock_registry.values.return_value = [MockImporter]
+        data = {"purl_string": "pkg:pypi/django@3.2"}
+        response = self.client.post(self.url, data, format="json")
+        assert response.status_code == 202
+        assert isinstance(response.data, list)
+        assert response.data[0]["importer"] == "dummy"
+        assert response.data[0]["purl"] == "pkg:pypi/django@3.2"
+        assert "steps_completed" in response.data[0]
+        assert "import" in response.data[0]["steps_completed"]
+
+    @patch("vulnerabilities.api_v2.LIVE_IMPORTERS_REGISTRY")
+    def test_evaluate_no_importer_found(self, mock_registry):
+        class MockImporter:
+            pipeline_id = "dummy"
+            supported_types = ["npm"]
+
+        mock_registry.values.return_value = [MockImporter]
+        data = {"purl_string": "pkg:pypi/django@3.2"}
+        response = self.client.post(self.url, data, format="json")
+        assert response.status_code == 400
+        assert "No live importers found" in response.data["error"]
+
+    def test_evaluate_invalid_purl(self):
+        data = {"purl_string": "not_a_valid_purl"}
+        response = self.client.post(self.url, data, format="json")
+        assert response.status_code == 400
+        assert "Invalid PackageURL" in response.data["error"]
+
+    @patch("vulnerabilities.api_v2.LIVE_IMPORTERS_REGISTRY")
+    def test_evaluate_no_threading(self, mock_registry):
+        class MockImporter:
+            pipeline_id = "dummy"
+            supported_types = ["pypi"]
+
+            def __init__(self, purl=None):
+                pass
+
+            def execute(self):
+                return 0, None
+
+        mock_registry.values.return_value = [MockImporter]
+        data = {"purl_string": "pkg:pypi/django@3.2", "no_threading": True}
+        response = self.client.post(self.url, data, format="json")
+        assert response.status_code == 202
+        assert isinstance(response.data, list)
+        assert response.data[0]["importer"] == "dummy"

vulnerablecode/urls.py

@@ -23,6 +23,7 @@
 from vulnerabilities.api_v2 import AdvisoriesPackageV2ViewSet
 from vulnerabilities.api_v2 import CodeFixV2ViewSet
 from vulnerabilities.api_v2 import CodeFixViewSet
+from vulnerabilities.api_v2 import LiveEvaluationViewSet
 from vulnerabilities.api_v2 import PackageV2ViewSet
 from vulnerabilities.api_v2 import PipelineScheduleV2ViewSet
 from vulnerabilities.api_v2 import VulnerabilityV2ViewSet
@@ -69,6 +70,7 @@ def __init__(self, *args, **kwargs):
 api_v2_router.register("codefixes", CodeFixViewSet, basename="codefix")
 api_v2_router.register("pipelines", PipelineScheduleV2ViewSet, basename="pipelines")
 api_v2_router.register("advisory-codefixes", CodeFixV2ViewSet, basename="advisory-codefix")
+api_v2_router.register("live-evaluation", LiveEvaluationViewSet, basename="live-evaluation")
 
 
 urlpatterns = [