Add support for reachability tools CPG using the integration layer

docs/index.rst

@@ -49,6 +49,7 @@ Learn via practical step-by-step guides.
 - :ref:`tutorial_cli_end_to_end_scanning_to_dejacode`
 - :ref:`tutorial_analyze_symbols_reachability`
 - :ref:`tutorial_reachability_integration_tools`
+- :ref:`tutorial_cpg_reachability_integration_tools`
 - :ref:`tutorial_web_ui_scan_repo_health`
 
 .. rst-class:: column column2 bottom-left
@@ -121,6 +122,7 @@ Indices and tables
     tutorial_cli_end_to_end_scanning_to_dejacode
     tutorial_analyze_symbols_reachability
     tutorial_reachability_integration_tools
+    tutorial_cpg_reachability_integration_tools
     tutorial_web_ui_scan_repo_health
     scanpipe-concepts
     built-in-pipelines

docs/tutorial_cpg_reachability_integration_tools.rst

@@ -0,0 +1,107 @@
+.. _tutorial_cpg_reachability_integration_tools:
+
+CPG Reachability Tools
+======================
+
+The ``cpg_reachability`` pipeline uses a
+`Code Property Graph (CPG) <https://github.com/Fraunhofer-AISEC/cpg>`_ to
+determine whether code symbols referenced by a vulnerability patch are
+reachable from a public entry point in your codebase. Each advisory is
+classified as ``yes``, ``unknown``, or ``no``.
+
+Requirements
+------------
+
+- `Fraunhofer CPG tool <https://github.com/Fraunhofer-AISEC/cpg>`_ installed and configured.
+- A running Neo4j instance with the APOC plugin:
+
+.. code-block:: console
+
+    docker run -d --name neo4j \
+      -p 7474:7474 -p 7687:7687 \
+      -e NEO4J_AUTH=neo4j/password \
+      -e NEO4J_PLUGINS='["apoc"]' \
+      -e NEO4J_server_memory_heap_max__size=6g \
+      -e NEO4J_server_memory_pagecache_size=1g \
+      neo4j:5
+
+Configuration
+-------------
+
+Set these environment variables in your ``.env``:
+
+.. code-block:: bash
+
+    CPG_NEO4J_HOST="localhost"       # Neo4j host
+    CPG_NEO4J_PORT="7687"            # Neo4j port
+    CPG_NEO4J_USER="neo4j"            # Neo4j username
+    CPG_NEO4J_PASSWORD="password"    # Neo4j password
+    CPG_NEO4J_EXECUTABLE="./cpg/cpg-neo4j/build/install/cpg-neo4j/bin/cpg-neo4j"  # Path to the ``cpg-neo4j`` executable
+
+Running the pipeline
+--------------------
+
+1. Open a Project that contains resources and at least one package
+   vulnerability.
+2. Click **"Add pipeline"** and select **"cpg_reachability"**.
+3. Select **"Execute pipeline now"** and click **"Add pipeline"**.
+4. After completion, open the **Resources** view (click the count under
+   **"RESOURCES"**), then click an affected file and open the **Extra**
+   tab to view the ``symbols_reachability`` data.
+
+.. image:: images/tutorial-web-ui-analyze-symbol-reachability.png
+
+Output
+------
+
+The pipeline writes two timestamped JSON reports and embeds the same
+advisory data in each resource's ``extra_data.symbols_reachability``.
+
+1. ``reachability-YYYY-MM-DD-HH-MM-SS.json`` the advisory reachability
+   report. Aggregates results per advisory, listing each
+   advisory with its overall verdict, matched vulnerable/fixed symbols,
+   and per-symbol witness paths.
+   Schema:
+
+.. code-block:: json
+    :emphasize-lines: 2, 7, 9, 13
+
+    {
+      "purl": "pkg:pypi/b2sdk@1.14.0",
+      "advisories": [
+        {
+          "advisory_uid": "pypa/b2sdk/PYSEC-2022-33",
+          "is_reachable": "yes",
+          "details": [
+            {
+              "symbol_name": "SqliteAccountInfo._create_database",
+              "is_defined": true,
+              "is_reachable": true,
+              "eog_path": [
+                "b2sdk.scan_folder",
+                "b2sdk.LocalSyncFolder._get_folder",
+                "SqliteAccountInfo._create_database"
+              ]
+            }
+          ],
+          "vulnerable_symbols": [
+            "SqliteAccountInfo",
+            "SqliteAccountInfo._create_database"
+          ],
+          "fixed_symbols": [
+            "SqliteAccountInfo",
+            "SqliteAccountInfo._create_database"
+          ]
+        }
+      ]
+    }
+
+2. ``cpg_reachability-YYYY-MM-DD-HH-MM-SS.json`` the graph
+   generated from CPG tool, serialized as a node/edge document.
+
+.. code-block:: text
+
+    {
+       "nodes": [...],
+       "edges": [...]
+    }

pyproject.toml

@@ -120,6 +120,8 @@ dependencies = [
   "stevedore==5.9.1",
   "sqlparse==0.6.0",
   "pluggy==1.6.0",
+  # Required to run the query (CPG reachability tool).
+  "neo4j==6.4.1",
 ]
 
 [project.optional-dependencies]
@@ -156,6 +158,7 @@ analyze_docker_image = "scanpipe.pipelines.analyze_docker:Docker"
 analyze_root_filesystem_or_vm_image = "scanpipe.pipelines.analyze_root_filesystem:RootFS"
 analyze_windows_docker_image = "scanpipe.pipelines.analyze_docker_windows:DockerWindows"
 analyze_symbols_reachability = "scanpipe.pipelines.analyze_symbols_reachability:SymbolReachability"
+cpg_symbols_reachability = "scanpipe.pipelines.cpg_symbols_reachability:CPGReachability"
 benchmark_purls = "scanpipe.pipelines.benchmark_purls:BenchmarkPurls"
 collect_strings_gettext = "scanpipe.pipelines.collect_strings_gettext:CollectStringsGettext"
 collect_symbols_ctags = "scanpipe.pipelines.collect_symbols_ctags:CollectSymbolsCtags"

scanpipe/pipelines/cpg_symbols_reachability.py

@@ -0,0 +1,137 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/aboutcode-org/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https://github.com/aboutcode-org/scancode.io for support and download.
+import subprocess
+from os import environ
+
+from scanpipe.pipes import cpg
+from scanpipe.pipes import run_command_safely
+from scanpipe.pipes.reachability_tools import ReachabilityPipeline
+from scanpipe.pipes.reachability_tools import ReachabilityTool
+
+CPG_NEO4J_EXECUTABLE = environ.get("CPG_NEO4J_EXECUTABLE")
+CPG_NEO4J_HOST = environ.get("CPG_NEO4J_HOST", "localhost")
+CPG_NEO4J_PORT = environ.get("CPG_NEO4J_PORT", "7687")
+CPG_NEO4J_USER = environ.get("CPG_NEO4J_USER", "neo4j")
+CPG_NEO4J_PASSWORD = environ.get("CPG_NEO4J_PASSWORD", "password")
+
+
+class CPGTool(ReachabilityTool):
+    """Reachability analysis tool based on a Code Property Graph (CPG)"""
+
+    tool_name = "cpg_tool"
+    executable = CPG_NEO4J_EXECUTABLE
+    supported_languages = ("Python", "Java", "JavaScript", "C++")
+
+    @classmethod
+    def get_availability(cls):
+        if not cls.executable:
+            return "CPG is not configured."
+
+    @classmethod
+    def run(cls, project, logger=None):
+        """Run the CPG client on the codebase and return the exported JSON Path."""
+        target_path = project.codebase_path
+        if not target_path:
+            resources = project.codebaseresources.all()
+            if not resources:
+                raise ValueError("No codebase resources found for this project.")
+            target_path = resources[0].location_path
+
+        export_json_path = project.get_output_file_path("cpg_reachability", "json")
+        command_args = [
+            CPG_NEO4J_EXECUTABLE,
+            "--host",
+            CPG_NEO4J_HOST,
+            "--port",
+            CPG_NEO4J_PORT,
+            "--user",
+            CPG_NEO4J_USER,
+            "--password",
+            CPG_NEO4J_PASSWORD,
+            "--export-json",
+            str(export_json_path),
+            "--top-level",
+            str(target_path),
+            str(target_path),
+        ]
+
+        logger(f"Generating CPG JSON for {target_path} ...")
+        try:
+            run_command_safely(command_args=command_args)
+            logger("CPG resource_index pipeline completed successfully")
+            return export_json_path
+        except subprocess.SubprocessError as error:
+            raise RuntimeError(f"CPG client failure: {error!r}")
+        except FileNotFoundError:
+            raise FileNotFoundError(
+                "CPG not found. Please ensure CPG is correctly configured."
+            )
+
+
+class CPGReachability(ReachabilityPipeline):
+    """
+    Determine the reachability of vulnerabilities using a Code Property
+    Graph (CPG) tool.
+
+    Note: You must run the ``find_vulnerabilities`` pipeline before running
+    this pipeline.
+
+    The project codebase is exported once as a CPG JSON graph. The vulnerable
+    and fixed symbols of each patch are matched against that graph, and a
+    symbol is reachable when it can be reached from a codebase entry point
+    following the evaluation order (EOG) of the graph. Results are stored in
+    the ``extra_data`` of the matching resources under the
+    ``symbols_reachability`` key, and a summary report is generated for each
+    vulnerability advisory.
+    """
+
+    reachability_tool = CPGTool
+
+    @classmethod
+    def get_availability(cls):
+        if not CPG_NEO4J_EXECUTABLE:
+            return "CPG is not configured."
+
+    def get_candidate_resources(self):
+        """Collect candidate resources your pipeline needs."""
+        self.candidate_resources = self.project.codebaseresources.files().filter(
+            is_binary=False,
+            is_archive=False,
+            is_media=False,
+            programming_language__in=self.reachability_tool.supported_languages,
+        )
+
+    def collect_resource_index(self):
+        """Collect resources symbols for each resource"""
+        self.resource_indexes_path = self.reachability_tool.run(
+            project=self.project, logger=self.log
+        )
+
+    def collect_and_match_resources(self):
+        """Match resource symbols against patch symbols."""
+        cpg.match_patches_to_resources(
+            tool_name=self.reachability_tool.tool_name,
+            patches=self.patches,
+            patch_symbols=self.patch_symbols,
+            candidate_resources=self.candidate_resources,
+            logger=self.log,
+        )