Update function docs

Add resource filter based on language we support

Signed-off-by: ziad hany <ziadhany2016@gmail.com>

Author: ziadhany

scanpipe/pipelines/analyze_symbols_reachability.py

@@ -22,23 +22,24 @@
 
 from scanpipe.pipelines import Pipeline
 from scanpipe.pipes import reachability
+from scanpipe.pipes.symbols import TS_QUERIES
 
 
 class SymbolReachability(Pipeline):
     """
-    Patch reachability analysis for given vulnerability patches.
+    Determine the reachability of vulnerabilities identified in the project.
 
-    For every patch we clone the git repository and extract
-    the vulnerable and fixed symbols from the patch commit.
+    Note: You must run `find_vulnerabilities` pipeline before running this pipeline.
 
-    These symbols are then matched against the project's codebase resources to
-    determine if the vulnerable code is actually present and reachable.
+    For every patch the git repository is cloned and extract the vulnerable and fixed
+    symbols from the patch commit. These symbols are then matched against
+    the project's codebase resources to determine if the vulnerable code
+    is actually present and reachable.
 
-    The analysis checks if vulnerable symbols are defined, imported, called, or
-    exactly match a fingerprint within the project files. The results, including
-    tool_details and a reachability status (REACHABLE, UNKNOWN, or
-    NOT_REACHABLE), are stored in the `extra_data` of the matching resources
-    under the `symbols_reachability` key.
+    The analysis checks if vulnerable symbols are defined, imported, called,
+    or exactly match a code within the project files. The results, including
+    tool_details and a reachability status (yes, unknown, or no), are stored
+    in the `extra_data` of the matching resources under the `symbols_reachability` key.
 
     Finally, a summary report is generated for each vulnerability
     advisory and saved as a JSON output file.
@@ -66,24 +67,22 @@ def get_vulnerabilities_patches(self):
         )
 
     def collect_resource_index(self):
-        """Collect resources symbols for each resource exactly once."""
+        """Collect resources symbols for each resource"""
         self.candidate_resources = self.project.codebaseresources.files().filter(
-            is_binary=False, is_archive=False, is_media=False
+            is_binary=False, is_archive=False, is_media=False,
+            programming_language__in=TS_QUERIES.keys()
         )
         self.resource_indexes = reachability.collect_resource_index(
             candidate_resources=self.candidate_resources
         )
 
     def collect_patch_symbols(self):
-        """
-        Collect patch symbols for all related commits,
-        then remove the local clone before moving on
-        """
+        """Collect patch symbols for all related commits."""
         self.patch_symbols = reachability.collect_patch_symbols(patches=self.patches)
 
     def collect_and_match_resources(self):
         """Match resource symbols against patch symbols."""
-        reachability.collect_and_match_resources(
+        reachability.match_patches_to_resources(
             patches=self.patches,
             patch_symbols=self.patch_symbols,
             resource_indexes=self.resource_indexes,

scanpipe/pipes/reachability.py

@@ -29,7 +29,7 @@
 
 from git import Repo
 from git.diff import NULL_TREE
-from scancode.api import get_file_info
+from typecode import get_type
 
 from scanpipe.pipes.symbols import TS_QUERIES
 from scanpipe.pipes.symbols import SymbolExtractor
@@ -55,10 +55,7 @@ def normalize_text(content):
 
 
 def detect_language_with_scancode(file_path, content):
-    """
-    Detect the programming language of the text
-    content using get_file_info function
-    """
+    """Detect the programming language of the text"""
     content = normalize_text(content)
 
     if not content:
@@ -67,11 +64,11 @@ def detect_language_with_scancode(file_path, content):
     tmp_dir = tempfile.mkdtemp(prefix="patch-lang-")
 
     try:
-        target = Path(tmp_dir) / Path(file_path).name
-        target.write_text(content, encoding="utf-8", errors="replace")
+        location = Path(tmp_dir) / Path(file_path).name
+        location.write_text(content, encoding="utf-8", errors="replace")
 
-        info = get_file_info(location=str(target)) or {}
-        return info.get("programming_language") or None
+        info = get_type(location)
+        return info.programming_language or None
 
     finally:
         shutil.rmtree(tmp_dir, ignore_errors=True)
@@ -652,7 +649,7 @@ def get_vulnerabilities_patches(package_vulnerabilities, dependency_vulnerabilit
 
 
 def collect_resource_index(candidate_resources):
-    """Collect resources symbols for each resource exactly once."""
+    """Collect resources symbols for each resource"""
     resource_indexes = {}
     for resource in candidate_resources:
         resource_language = resource.programming_language
@@ -674,8 +671,8 @@ def collect_resource_index(candidate_resources):
 
 def collect_patch_symbols(patches):
     """
-    For each unique repo: clone it once, collect patch symbols for all
-    related commits, then remove the local clone before moving on
+    For each unique repo clone it once,
+    collect patch symbols for all related commits
     """
     patch_symbols = {}
     patches_by_repo = {}
@@ -704,7 +701,7 @@ def collect_patch_symbols(patches):
     return patch_symbols
 
 
-def collect_and_match_resources(
+def match_patches_to_resources(
     patches, patch_symbols, candidate_resources, resource_indexes
 ):
     """Match resource symbols against patch symbols."""
@@ -760,18 +757,18 @@ def collect_and_match_resources(
 
 def generate_advisory_reachability_report(project, patches, candidate_resources):
     """
-    Generate reachability report keyed by advisory_uid.
+    Generate a reachability report summarizing status by advisory.
 
     Each advisory contains its overall reachability status
     and the reachability results collected from all resources
     and associated patches.
 
     The overall reachability status is determined using the following
-    priority order: REACHABLE > UNKNOWN > NOT_REACHABLE
+    priority order: REACHABLE: "yes" > UNKNOWN: "unknown" > NOT_REACHABLE: "no"
 
     This means that an advisory is considered reachable if it is reachable
     through at least one resource or patch. If no reachable result exists,
-    but at least one result is unknown, the advisory status is UNKNOWN.
+    but at least one result is UNKNOWN, the advisory status is UNKNOWN.
     Otherwise, it is NOT_REACHABLE.
     """
     status_priority = {