Update the pipeline to have a test for resource_patch_matcher

Refactor the pipeline

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

Author: ziadhany

scanpipe/pipelines/analyze_symbols_reachability.py

@@ -47,7 +47,7 @@ class SymbolReachability(Pipeline):
 
     The analysis checks if vulnerable symbols are defined, imported, called, or
     exactly match a fingerprint within the project files. The results, including
-    evidence and a reachability status (REACHABLE, UNKNOWN, or
+    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.
 
@@ -176,10 +176,10 @@ def collect_and_match_resources(self):
                 fixed_symbols = patch_symbols.get("fixed", {})
 
                 matcher = ResourcePatchMatcher(resource_index=resource_index)
-                vuln_evidence = matcher.match(vulnerable_symbols)
-                fixed_evidence = matcher.match(fixed_symbols)
+                vuln_details = matcher.match(vulnerable_symbols)
+                fixed_details = matcher.match(fixed_symbols)
 
-                if not any([vuln_evidence, fixed_evidence]):
+                if not any([vuln_details, fixed_details]):
                     continue
 
                 report = {
@@ -188,10 +188,10 @@ def collect_and_match_resources(self):
                         "commit_hash": commit_hash,
                     },
                     "advisory_uids": advisory_uids,
-                    "evidence": list(vuln_evidence.values()),
-                    "fixed_symbols": sorted(fixed_evidence.keys()),
-                    "vulnerable_symbols": sorted(vuln_evidence.keys()),
-                    "reachability_status": classify_reachability(vuln_evidence).value,
+                    "tool_details": list(vuln_details.values()),
+                    "fixed_symbols": sorted(fixed_details.keys()),
+                    "vulnerable_symbols": sorted(vuln_details.keys()),
+                    "is_reachable": classify_reachability(vuln_details).value,
                 }
 
                 add_reachability_report(
@@ -223,40 +223,44 @@ def generate_advisory_reachability_report(self):
             ReachabilityStatus.NOT_REACHABLE.value: 1,
         }
 
-        advisory_reachability_report = {}
+        advisory_reachability_report = {
+            "purl": self.project.purl,
+            "advisories": [],
+        }
+
+        advisory_map = {}
         for resource in self.candidate_resources:
             for report in resource.extra_data.get("symbols_reachability", []):
                 advisory_uids = report.get("advisory_uids", [])
-                reachability_status = report.get("reachability_status")
+                is_reachable = report.get("is_reachable")
                 patch = report.get("patch", {})
 
                 for adv_uid in advisory_uids:
-                    if adv_uid not in advisory_reachability_report:
-                        advisory_reachability_report[adv_uid] = {
-                            "reachable": ReachabilityStatus.NOT_REACHABLE.value,
+                    if adv_uid not in advisory_map:
+                        adv_data = {
+                            "advisory_uid": adv_uid,
+                            "is_reachable": ReachabilityStatus.NOT_REACHABLE.value,
                             "details": [],
                         }
+                        advisory_map[adv_uid] = adv_data
+                        advisory_reachability_report["advisories"].append(adv_data)
 
                     tool_details = {
                         "resource_path": resource.path,
                         "patch": patch,
-                        "reachability_status": reachability_status,
-                        "evidence": report.get("evidence", []),
+                        "is_reachable": is_reachable,
+                        "tool_details": report.get("tool_details", []),
                         "vulnerable_symbols": report.get("vulnerable_symbols", []),
                         "fixed_symbols": report.get("fixed_symbols", []),
                     }
 
-                    advisory_reachability_report[adv_uid]["details"].append(
-                        tool_details
-                    )
+                    advisory_map[adv_uid]["details"].append(tool_details)
 
-                    current_status = advisory_reachability_report[adv_uid]["reachable"]
-                    if status_priority.get(
-                        reachability_status, 0
-                    ) > status_priority.get(current_status, 0):
-                        advisory_reachability_report[adv_uid]["reachable"] = (
-                            reachability_status
-                        )
+                    current_status = advisory_map[adv_uid]["is_reachable"]
+                    if status_priority.get(is_reachable, 0) > status_priority.get(
+                        current_status, 0
+                    ):
+                        advisory_map[adv_uid]["is_reachable"] = is_reachable
 
         reachability_output_path = self.project.get_output_file_path(
             "reachability", "json"

scanpipe/pipes/reachability.py

@@ -36,9 +36,9 @@
 
 
 class ReachabilityStatus(str, Enum):
-    REACHABLE = "YES"
-    UNKNOWN = "UNKNOWN"
-    NOT_REACHABLE = "NO"
+    REACHABLE = "yes"
+    UNKNOWN = "unknown"
+    NOT_REACHABLE = "no"
 
 
 def normalize_text(content):
@@ -347,21 +347,21 @@ def analyze(
         return vuln_meta, fixed_meta, language
 
 
-def classify_reachability(evidence):
+def classify_reachability(tool_details):
     """
     Classify the reachability status of a vulnerability based on the
-    collected evidence from :class:`ResourcePatchMatcher`.
+    collected tool_details from ResourcePatchMatcher.
     """
-    if not evidence:
+    if not tool_details:
         return ReachabilityStatus.NOT_REACHABLE
 
     status = ReachabilityStatus.NOT_REACHABLE
-    for item in evidence.values():
-        is_called = bool(item.get("called"))
+    for item in tool_details.values():
+        is_called = bool(item.get("is_called"))
         has_path = bool(item.get("reachable_from"))
-        is_defined = bool(item.get("defined"))
-        is_imported = bool(item.get("imported"))
-        is_exact = bool(item.get("fingerprint"))
+        is_defined = bool(item.get("is_defined"))
+        is_imported = bool(item.get("is_imported"))
+        is_exact = bool(item.get("is_exact"))
 
         if is_exact or (is_imported and (is_called or has_path)):
             return ReachabilityStatus.REACHABLE
@@ -461,16 +461,68 @@ def __init__(self, resource_index):
         self.imports = resource_index.get("imports", {})
         self.callers_of = resource_index.get("callers_of", {})
         self.separator = resource_index.get("separator", ".")
+        self.wildcard_modules = self.imports.get("*", [])
+
+    def _matches_first_component(
+        self, qualified_name, abs_path, local_name, import_call_names
+    ):
+        """
+        Check if the first component of qualified_name
+        matches the end of abs_path.
+        """
+        first_component = qualified_name.split(self.separator, 1)[0]
+        if first_component != qualified_name and (
+            abs_path.endswith(self.separator + first_component)
+            or abs_path == first_component
+        ):
+            remaining = qualified_name[len(first_component) :]
+            import_call_names.add(f"{local_name}{remaining}")
+            return True
+        return False
+
+    def _matches_wildcard(self, qualified_name):
+        """Check if the qualified_name is covered by a wildcard import."""
+        return any(
+            qualified_name == mod or qualified_name.startswith(mod + self.separator)
+            for mod in self.wildcard_modules
+        )
+
+    def _get_import_info(self, qualified_name):
+        """Check if qualified_name is imported and return possible call names."""
+        import_call_names = set()
+        imported = False
+
+        for local_name, abs_path in self.imports.items():
+            if local_name == "*":
+                continue
+
+            if qualified_name in (local_name, abs_path):
+                imported = True
+                import_call_names.add(local_name)
+            elif qualified_name.startswith(local_name + self.separator):
+                imported = True
+                import_call_names.add(qualified_name)
+            elif qualified_name.startswith(abs_path + self.separator):
+                imported = True
+                remaining = qualified_name[len(abs_path) :]
+                import_call_names.add(f"{local_name}{remaining}")
+            elif abs_path.endswith(self.separator + qualified_name):
+                imported = True
+                import_call_names.add(local_name)
+            elif self._matches_first_component(
+                qualified_name, abs_path, local_name, import_call_names
+            ):
+                imported = True
+
+        if not imported and self._matches_wildcard(qualified_name):
+            return True, import_call_names
+
+        return imported, import_call_names
 
     def match(self, patch_symbols_metadata):
         """
         Match a set of patch symbols against the resource index and
-        return evidence for each matched symbol.
-
-        For each symbol in patch_symbols_metadata, the method
-        checks whether it is defined, imported, called, or has an
-        exact fingerprint match in the resource. If at least one of
-        these conditions is true, an evidence entry is created.
+        return tool_details for each matched symbol.
         """
         if not patch_symbols_metadata or not self.resource_index:
             return {}
@@ -480,42 +532,56 @@ def match(self, patch_symbols_metadata):
             qualified_name = metadata["qualified_name"]
             fingerprint = metadata["fingerprint"]
             defined = qualified_name in self.definitions
-            fingerprint_hit = bool(fingerprint and fingerprint in self.fingerprints)
-
-            imported = (
-                qualified_name in self.imports
-                or qualified_name in self.imports.values()
+            is_exact = bool(
+                fingerprint
+                and fingerprint in self.fingerprints
+                and qualified_name in self.definitions
             )
+            short_name = (
+                qualified_name.rsplit(self.separator, 1)[-1]
+                if self.separator in qualified_name
+                else qualified_name
+            )
+
+            imported, import_call_names = self._get_import_info(qualified_name)
 
-            callers = set(self.callers_of.get(qualified_name, set()))
-            called = bool(callers)
+            possible_call_names = {qualified_name}
+            if imported or defined:
+                possible_call_names.add(short_name)
+            possible_call_names.update(import_call_names)
 
-            if not (defined or fingerprint_hit or called or imported):
+            callers = set()
+            for call_name in possible_call_names:
+                callers.update(self.callers_of.get(call_name, set()))
+
+            called = bool(callers) and (
+                imported or defined or bool(self.wildcard_modules)
+            )
+            if called and not imported and not defined and self.wildcard_modules:
+                imported = True
+
+            if not (defined or is_exact or called or imported):
                 continue
 
             entry = matched.setdefault(
                 qualified_name,
                 {
                     "symbol_name": qualified_name,
-                    "called": False,
-                    "defined": False,
-                    "imported": False,
-                    "fingerprint": None,
+                    "is_called": False,
+                    "is_defined": False,
+                    "is_imported": False,
+                    "is_exact": False,
                     "reachable_from": [],
                 },
             )
 
-            if defined:
-                entry["defined"] = True
-            if imported:
-                entry["imported"] = True
+            entry["is_defined"] = entry["is_defined"] or defined
+            entry["is_imported"] = entry["is_imported"] or imported
+            entry["is_exact"] = entry["is_exact"] or is_exact
+            entry["is_called"] = entry["is_called"] or called
             if called:
-                entry["called"] = True
                 entry["reachable_from"] = sorted(callers)
 
-            if fingerprint_hit:
-                entry["fingerprint"] = fingerprint
-
         return matched
 
 

scanpipe/pipes/symbols.py

@@ -377,7 +377,12 @@ class JavaTreeSitterQuery(LanguageQuery):
         (import_declaration (scoped_identifier) @import_name)
         (import_declaration (scoped_identifier) @module_name (asterisk) @import_name)
     """
-    syntax_config = {"self_keyword": "this", "separator": ".", "wildcard_symbol": "*"}
+    syntax_config = {
+        "self_keyword": "this",
+        "separator": ".",
+        "wildcard_symbol": "*",
+        "import_local_name": "last",
+    }
 
 
 TS_QUERIES = {
@@ -469,9 +474,29 @@ def extract_calls(self, node):
 
         return calls
 
+    def resolve_local_name(self, imp_name, alias):
+        """Resolve the local name for an imported symbol."""
+        separator = self.syntax_config.get("separator", ".")
+        local_name = alias or imp_name
+
+        if not alias and separator in imp_name:
+            if self.syntax_config.get("import_local_name") == "last":
+                return imp_name.split(separator)[-1]
+            return imp_name.split(separator)[0]
+
+        return local_name
+
+    def resolve_absolute_path(self, module_name, imp_name):
+        """Resolve the absolute path for an imported symbol."""
+        separator = self.syntax_config.get("separator", ".")
+        if module_name:
+            if module_name == separator:
+                return f"{separator}{imp_name}"
+            return f"{module_name}{separator}{imp_name}"
+        return imp_name
+
     def extract_imports(self):
         """Map every local alias to its absolute imported path."""
-        separator = self.syntax_config.get("separator", ".")
         wildcard_sym = self.syntax_config.get("wildcard_symbol")
 
         import_map = {}
@@ -487,17 +512,8 @@ def extract_imports(self):
                         wildcard_modules.append(module_name)
                     continue
 
-                local_name = alias or imp_name
-                if not alias and separator in imp_name:
-                    local_name = imp_name.split(separator)[0]
-
-                if module_name:
-                    if module_name == separator:
-                        absolute_path = f"{separator}{imp_name}"
-                    else:
-                        absolute_path = f"{module_name}{separator}{imp_name}"
-                else:
-                    absolute_path = imp_name
+                local_name = self.resolve_local_name(imp_name, alias)
+                absolute_path = self.resolve_absolute_path(module_name, imp_name)
 
                 import_map[local_name] = absolute_path