Add Support for CPG tool

docker.env

@@ -7,4 +7,9 @@ SCANCODEIO_DB_HOST=db
 SCANCODEIO_RQ_REDIS_HOST=redis
 SCANCODEIO_ASYNC=True
 SCANCODEIO_WORKSPACE_LOCATION=/var/scancodeio/workspace/
-ATOM_EXECUTABLE=/atom-root/bin/atom
+ATOM_EXECUTABLE=/atom-root/bin/atom
+CPG_NEO4J_EXECUTABLE="./cpg/cpg-neo4j/build/install/cpg-neo4j/bin/cpg-neo4j"
+CPG_NEO4J_HOST="localhost"
+CPG_NEO4J_PORT="7687"
+CPG_NEO4J_USER="neo4j"
+CPG_NEO4J_PASSWORD="password"

docs/source/index.rst

@@ -6,6 +6,7 @@ Welcome to ReachableCode's documentation!
    :caption: Contents:
 
    tutorial_atom_reachability_integration_tools
+   tutorial_cpg_reachability_integration_tools
    contribute/contrib_doc
 
 Indices and tables

docs/source/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-cpg-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": [...]
+    }

setup.cfg

@@ -44,6 +44,7 @@ python_requires = >=3.8
 
 install_requires =
     commoncode
+    neo4j
 
 
 [options.packages.find]
@@ -74,4 +75,5 @@ docs =
 
 [options.entry_points]
 scancodeio_pipelines =
-    analyze_atom_reachability = reachablecode.pipelines.analyze_atom_reachability:AtomReachability
+    analyze_atom_reachability = reachablecode.pipelines.analyze_atom_reachability:AtomReachability
+    analyze_cpg_reachability = reachablecode.pipelines.analyze_cpg_reachability:CPGReachability

src/reachablecode/pipelines/analyze_cpg_reachability.py

@@ -0,0 +1,79 @@
+# 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.
+
+
+from scanpipe.pipes.reachability_tools import ReachabilityPipeline
+
+from reachablecode.pipes import cpg
+from reachablecode.pipes.cpg import CPGTool
+from os import environ
+
+CPG_NEO4J_EXECUTABLE = environ.get("CPG_NEO4J_EXECUTABLE")
+
+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,
+        )