Add changelog entry and unit test for check_vulnerabilities #101

Signed-off-by: Thomas Druez <[email protected]>

Author: tdruez

CHANGELOG.rst

@@ -4,6 +4,12 @@ Changelog
 v31.1.0 (unreleased)
 --------------------
 
+- Add a new "check vulnerabilities" pipeline to lookup vulnerabilities in the
+  VulnerableCode database for all project discovered packages.
+  Vulnerability data is stored in the extra_data field of each package.
+  More details about VulnerableCode at https://github.com/nexB/vulnerablecode/
+  https://github.com/nexB/scancode.io/issues/101
+
 - Generate SBOM (Software Bill of Materials) compliant with the SPDX 2.3 specification
   as a new downloadable output.
   https://github.com/nexB/scancode.io/issues/389

scanpipe/pipes/vulnerablecode.py

@@ -58,32 +58,40 @@ def is_configured():
     return False
 
 
-def is_service_available(label, session, url, raise_exceptions):
+def is_available():
     """
-    Base function that checks if a configured integration service is available.
+    Returns True if the configured VulnerableCode server is available.
     """
+    if not is_configured():
+        return False
+
     try:
-        response = session.head(url)
+        response = session.head(VULNERABLECODE_API_URL)
         response.raise_for_status()
     except requests.exceptions.RequestException as request_exception:
         logger.debug(f"{label} is_available() error: {request_exception}")
-        if raise_exceptions:
-            raise
         return False
 
     return response.status_code == requests.codes.ok
 
 
-def is_available(raise_exceptions=False):
+def get_base_purl(purl):
     """
-    Returns True if the configured VulnerableCode server is available.
+    Returns the `purl` without qualifiers and subpath.
     """
-    if not is_configured():
-        return False
+    return purl.split("?")[0]
 
-    return is_service_available(
-        label, session, VULNERABLECODE_API_URL, raise_exceptions
-    )
+
+def get_purls(packages, base=False):
+    """
+    Returns the PURLs for the given list of `packages`.
+    Do not include qualifiers nor subpath when `base` is provided.
+    """
+    return [
+        get_base_purl(package_url) if base else package_url
+        for package in packages
+        if (package_url := package.package_url)
+    ]
 
 
 def request_get(
@@ -123,13 +131,6 @@ def request_post(
         logger.debug(f"{label} [Exception] {exception}")
 
 
-def get_base_purl(purl):
-    """
-    Returns the `purl` without the qualifiers and the subpath.
-    """
-    return purl.split("?")[0]
-
-
 def _get_vulnerabilities(
     url,
     field_name,
@@ -213,62 +214,3 @@ def bulk_search_by_cpes(
 
     logger.debug(f"VulnerableCode: url={url} cpes_count={len(cpes)}")
     return request_post(url, data, timeout)
-
-
-def get_purls(packages):
-    """
-    Returns the PURLs for the given list of `packages`.
-    List comprehension is not used on purpose to avoid crafting each
-    PURL twice.
-    """
-    purls = []
-    for package in packages:
-        package_url = package.package_url
-        if package_url:
-            purls.append(package_url)
-    return purls
-
-
-def get_vulnerable_purls(packages):
-    """
-    Returns a list of PURLs for which at least one `affected_by_vulnerabilities`
-    was found in the VulnerableCodeDB for the given list of `packages`.
-    """
-    purls = get_purls(packages)
-
-    if not purls:
-        return []
-
-    search_results = bulk_search_by_purl(purls, timeout=5)
-    if not search_results:
-        return []
-
-    return [
-        entry.get("purl")
-        for entry in search_results
-        if entry.get("affected_by_vulnerabilities")
-    ]
-
-
-def get_vulnerable_cpes(components):
-    """
-    Returns a list of vulnerable CPEs found in the VulnerableCodeDB for the given list
-    of `components`.
-    """
-    cpes = [component.cpe for component in components if component.cpe]
-
-    if not cpes:
-        return []
-
-    search_results = bulk_search_by_cpes(cpes, timeout=5)
-    if not search_results:
-        return []
-
-    vulnerable_cpes = [
-        reference.get("reference_id")
-        for entry in search_results
-        for reference in entry.get("references")
-        if reference.get("reference_id").startswith("cpe")
-    ]
-
-    return list(set(vulnerable_cpes))

scanpipe/tests/test_pipelines.py

@@ -35,12 +35,14 @@
 
 from scancode.cli_test_utils import purl_with_fake_uuid
 
+from scanpipe.models import DiscoveredPackage
 from scanpipe.models import Project
 from scanpipe.pipelines import Pipeline
 from scanpipe.pipelines import is_pipeline
 from scanpipe.pipelines import root_filesystems
 from scanpipe.pipes import output
 from scanpipe.tests import FIXTURES_REGEN
+from scanpipe.tests import package_data1
 from scanpipe.tests.pipelines.do_nothing import DoNothing
 from scanpipe.tests.pipelines.steps_as_attribute import StepsAsAttribute
 
@@ -580,3 +582,45 @@ def test_scanpipe_load_inventory_pipeline_integration_test(self):
             self.data_location / "asgiref-3.3.0_load_inventory_expected.json"
         )
         self.assertPipelineResultEqual(expected_file, result_file)
+
+    @mock.patch("scanpipe.pipes.vulnerablecode.is_available")
+    @mock.patch("scanpipe.pipes.vulnerablecode.is_configured")
+    @mock.patch("scanpipe.pipes.vulnerablecode.get_vulnerabilities_by_purl")
+    def test_scanpipe_check_vulnerabilities_pipeline_integration_test(
+        self, mock_get_vulnerabilities, mock_is_configured, mock_is_available
+    ):
+        pipeline_name = "check_vulnerabilities"
+        project1 = Project.objects.create(name="Analysis")
+        package1 = DiscoveredPackage.create_from_data(project1, package_data1)
+
+        run = project1.add_pipeline(pipeline_name)
+        pipeline = run.make_pipeline_instance()
+        mock_is_configured.return_value = False
+        mock_is_available.return_value = False
+        exitcode, out = pipeline.execute()
+        self.assertEqual(1, exitcode, msg=out)
+        self.assertIn("VulnerableCode is not configured.", out)
+
+        run = project1.add_pipeline(pipeline_name)
+        pipeline = run.make_pipeline_instance()
+        mock_is_configured.return_value = True
+        mock_is_available.return_value = True
+        vulnerability_data = [
+            {
+                "purl": "pkg:deb/debian/[email protected]",
+                "affected_by_vulnerabilities": [
+                    {
+                        "vulnerability_id": "VCID-cah8-awtr-aaad",
+                        "summary": "An issue was discovered.",
+                    }
+                ],
+            }
+        ]
+        mock_get_vulnerabilities.return_value = vulnerability_data
+
+        exitcode, out = pipeline.execute()
+        self.assertEqual(0, exitcode, msg=out)
+
+        package1.refresh_from_db()
+        expected = {"discovered_vulnerabilities": vulnerability_data}
+        self.assertEqual(expected, package1.extra_data)