Import vulnerability data from ScanCode.io

product_portfolio/importers.py

@@ -699,9 +699,33 @@ def import_dependencies(self):
         for dependency_data in self.dependencies:
             self.import_dependency(dependency_data)
 
+    @staticmethod
+    def import_vulnerability(vulnerability_data, product_package):
+        from vulnerabilities.models import VulnerabilityAnalysis
+
+        vulnerability_id = vulnerability_data.get("vulnerability_id")
+        if not vulnerability_id:
+            return
+
+        package = product_package.package
+        # TODO: Add a lower-level create_vulnerability method
+        vulnerabilities = package.create_vulnerabilities(vulnerabilities_data=[vulnerability_data])
+        vulnerability = vulnerabilities[0]
+
+        if cdx_vulnerability := vulnerability_data.get("cdx_vulnerability"):
+            if analysis_data := cdx_vulnerability.get("analysis"):
+                VulnerabilityAnalysis.create_from_data(
+                    user=product_package.dataspace,
+                    data={
+                        "product_package": product_package,
+                        "vulnerability": vulnerability,
+                        **analysis_data,
+                    },
+                )
+
     def import_package(self, package_data):
-        # Vulnerabilities are fetched post import.
-        package_data.pop("affected_by_vulnerabilities", None)
+        # Vulnerabilities are assigned after the package creation.
+        affected_by_vulnerabilities = package_data.pop("affected_by_vulnerabilities", [])
 
         # Check if the package already exists to prevent duplication.
         package = self.look_for_existing_package(package_data)
@@ -730,7 +754,7 @@ def import_package(self, package_data):
                 return
             self.created["package"].append(str(package))
 
-        ProductPackage.objects.get_or_create(
+        product_package, _ = ProductPackage.objects.get_or_create(
             product=self.product,
             package=package,
             dataspace=self.product.dataspace,
@@ -743,6 +767,9 @@ def import_package(self, package_data):
         package_uid = package_data.get("package_uid") or package.uuid
         self.package_uid_mapping[package_uid] = package
 
+        for vulnerability_data in affected_by_vulnerabilities:
+            self.import_vulnerability(vulnerability_data, product_package)
+
     def import_dependency(self, dependency_data):
         dependency_uid = dependency_data.get("dependency_uid")
 

product_portfolio/tests/test_importers.py

@@ -1255,3 +1255,56 @@ def test_product_portfolio_import_packages_from_scio_importer_duplicate_dependen
         self.assertEqual({}, errors)
         self.assertEqual(2, self.product1.packages.count())
         self.assertEqual(1, self.product1.dependencies.count())
+
+    @mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_dependencies")
+    @mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_packages")
+    def test_product_portfolio_import_packages_from_scio_importer_vex(
+        self, mock_fetch_packages, mock_fetch_dependencies
+    ):
+        vulnerability_data = {
+            # "id": "VCID-0001",
+            "vulnerability_id": "VCID-0001",
+            "summary": "complexity bugs may lead to a denial of service",
+            "cdx_vulnerability": {
+                "affects": [{"ref": "pkg:maven/abc/abc@1.0"}],
+                "bom-ref": "BomRef.1",
+                "description": "complexity bugs may lead to a denial of service",
+                "analysis": {
+                    "detail": "AAAA",
+                    "justification": "code_not_present",
+                    "responses": ["can_not_fix", "update"],
+                    "state": "resolved",
+                },
+            },
+        }
+        mock_fetch_packages.return_value = [
+            {
+                "purl": "pkg:maven/abc/abc@1.0",
+                "type": "maven",
+                "namespace": "abc",
+                "name": "abc",
+                "version": "1.0",
+                "affected_by_vulnerabilities": [vulnerability_data],
+            }
+        ]
+
+        importer = ImportPackageFromScanCodeIO(
+            user=self.super_user,
+            project_uuid=uuid.uuid4(),
+            product=self.product1,
+            infer_download_urls=True,
+        )
+        created, existing, errors = importer.save()
+        created_package = self.product1.packages.get()
+        vulnerability = created_package.affected_by_vulnerabilities.get()
+        self.assertEqual(vulnerability_data["vulnerability_id"], vulnerability.vulnerability_id)
+        self.assertEqual(vulnerability_data["summary"], vulnerability.summary)
+
+        analysis = vulnerability.vulnerability_analyses.get()
+        self.assertEqual(vulnerability, analysis.vulnerability)
+        self.assertEqual(self.product1, analysis.product)
+        self.assertEqual(created_package, analysis.package)
+        self.assertEqual("resolved", analysis.state)
+        self.assertEqual("code_not_present", analysis.justification)
+        self.assertEqual("AAAA", analysis.detail)
+        self.assertEqual(["can_not_fix", "update"], analysis.responses)

vulnerabilities/models.py

@@ -487,6 +487,8 @@ def create_vulnerabilities(self, vulnerabilities_data):
         if isinstance(self, Package):
             self.productpackages.update_weighted_risk_score()
 
+        return vulnerabilities
+
 
 class VulnerabilityAnalysis(
     VulnerabilityAnalysisMixin,