diff --git a/pyproject.toml b/pyproject.toml index 350c59a0b7..b309d0a955 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -151,6 +151,7 @@ run = "scancodeio:combined_run" 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" 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" diff --git a/scanpipe/pipelines/analyze_symbols_reachability.py b/scanpipe/pipelines/analyze_symbols_reachability.py new file mode 100644 index 0000000000..4eaf4116e1 --- /dev/null +++ b/scanpipe/pipelines/analyze_symbols_reachability.py @@ -0,0 +1,116 @@ +# 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.pipelines import Pipeline +from scanpipe.pipes import reachability +from scanpipe.pipes.symbols import TS_QUERIES + + +class SymbolReachability(Pipeline): + """ + Determine the reachability of vulnerabilities identified in the project. + + Note: You must run `find_vulnerabilities` pipeline before running this pipeline. + + 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 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. + """ + + download_inputs = False + is_addon = True + results_url = "/project/{slug}/resources/?extra_data=symbol_reachability" + + @classmethod + def steps(cls): + return ( + cls.get_vulnerabilities_patches, + cls.collect_resource_index, + cls.collect_patch_symbols, + cls.collect_and_match_resources, + cls.generate_advisory_reachability_report, + cls.apply_reachability_to_packages_and_dependencies, + ) + + def get_vulnerabilities_patches(self): + """Get unique patch for all vulnerabilities.""" + self.patches = reachability.get_vulnerabilities_patches( + package_vulnerabilities=self.project.package_vulnerabilities, + dependency_vulnerabilities=self.project.dependency_vulnerabilities, + ) + + def collect_resource_index(self): + """Collect resources symbols for each resource""" + self.candidate_resources = self.project.codebaseresources.files().filter( + 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, logger=self.log + ) + + def collect_patch_symbols(self): + """Collect patch symbols for all related commits.""" + self.patch_symbols = reachability.collect_patch_symbols( + patches=self.patches, logger=self.log + ) + + def collect_and_match_resources(self): + """Match resource symbols against patch symbols.""" + reachability.match_patches_to_resources( + patches=self.patches, + patch_symbols=self.patch_symbols, + resource_indexes=self.resource_indexes, + candidate_resources=self.candidate_resources, + logger=self.log, + ) + + def generate_advisory_reachability_report(self): + """Generate a reachability report summarizing status by advisory.""" + self.advisories_reachability_report = ( + reachability.generate_advisory_reachability_report( + project=self.project, + patches=self.patches, + candidate_resources=self.candidate_resources, + ) + ) + + def apply_reachability_to_packages_and_dependencies(self): + """ + Save reachability results by updating DiscoveredPackage and + DiscoveredDependency records with the computed reachability data + in their affected_by_vulnerabilities JSON field. + """ + reachability.apply_reachability_to_packages_and_dependencies( + project=self.project, advisory_report=self.advisories_reachability_report + ) diff --git a/scanpipe/pipes/reachability.py b/scanpipe/pipes/reachability.py new file mode 100644 index 0000000000..ee0dc5993a --- /dev/null +++ b/scanpipe/pipes/reachability.py @@ -0,0 +1,894 @@ +# 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 difflib +import json +import shutil +import tempfile +from enum import Enum +from pathlib import Path + +from git import Repo +from git.diff import NULL_TREE +from typecode import get_type + +from aboutcode.pipeline import LoopProgress +from scanpipe.models import DiscoveredDependency +from scanpipe.models import DiscoveredPackage +from scanpipe.pipes.symbols import TS_QUERIES +from scanpipe.pipes.symbols import SymbolExtractor +from scanpipe.pipes.symbols import create_sha256_fingerprint +from scanpipe.pipes.symbols import is_supported_language + + +class ReachabilityStatus(str, Enum): + REACHABLE = "yes" + UNKNOWN = "unknown" + NOT_REACHABLE = "no" + + +def normalize_text(content): + """Normalize content (bytes) into a UTF-8 decoded string.""" + if content is None: + return "" + + if isinstance(content, bytes): + return content.decode("utf-8", errors="replace") + + return str(content) + + +def detect_language_with_scancode(file_path, content): + """Detect the programming language of the text""" + content = normalize_text(content) + + if not content: + return None + + tmp_dir = tempfile.mkdtemp(prefix="patch-lang-") + + try: + location = Path(tmp_dir) / Path(file_path).name + location.write_text(content, encoding="utf-8", errors="replace") + + info = get_type(location) + return info.programming_language or None + + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +class PatchAnalyzer: + def __init__(self, repo, commit_hash): + self.repo = repo + self.commit = repo.commit(commit_hash) + self.parent_commit = self.commit.parents[0] if self.commit.parents else None + + def get_changed_files(self): + """ + Retrieve all files changed by the commit along with their + vulnerable and fixed contents. + + For each changed file, a dictionary entry is created with two + keys: + + - vulnerable_text: The file content before the commit (empty + string for newly added files). + - fixed_text: The file content after the commit (empty string + for deleted files). + + """ + diffs = ( + self.parent_commit.diff(self.commit, create_patch=False) + if self.parent_commit + else self.commit.diff(NULL_TREE, create_patch=False) + ) + + files = {} + for diff in diffs: + change_type = diff.change_type + old_path = diff.a_path if change_type in ("D", "M", "R") else None + new_path = diff.b_path if change_type in ("A", "M", "R") else None + path_key = new_path or old_path + + if not path_key: + continue + + entry = files.setdefault( + path_key, {"vulnerable_text": "", "fixed_text": ""} + ) + + if old_path and self.parent_commit: + entry["vulnerable_text"] = ( + (self.parent_commit.tree / old_path) + .data_stream.read() + .decode("utf-8", errors="replace") + ) + + if new_path: + entry["fixed_text"] = ( + (self.commit.tree / new_path) + .data_stream.read() + .decode("utf-8", errors="replace") + ) + + return files + + def get_commit_diff_text(self): + """Get the diff text, falling back to an empty tree if no parent exists.""" + base = self.parent_commit.hexsha if self.parent_commit else NULL_TREE + return self.repo.git.diff(base, self.commit.hexsha, unified=3) + + @classmethod + def compute_changed_lines(cls, vulnerable_text, fixed_text): + """Return the removed and added line numbers between two file contents.""" + matcher = difflib.SequenceMatcher( + a=vulnerable_text.splitlines(), + b=fixed_text.splitlines(), + autojunk=False, # otherwise difflib ignores lines that repeat often + ) + + removed_lines = [] + added_lines = [] + for tag, vuln_start, vuln_end, fixed_start, fixed_end in matcher.get_opcodes(): + if tag == "equal": + continue + # opcodes are 0-based and end-exclusive, line numbers are 1-based + removed_lines.extend(range(vuln_start + 1, vuln_end + 1)) + added_lines.extend(range(fixed_start + 1, fixed_end + 1)) + + return removed_lines, added_lines + + @classmethod + def diff_changed_symbols(cls, vuln_meta, fixed_meta): + """ + Compare the vulnerable and fixed symbol metadata and return the + symbols that are unique to each side (i.e., whose body text + differs between the two versions). + + A symbol key is considered "vulnerable-only" if its body text + does not match the corresponding symbol in fixed_meta, and + vice versa. + """ + vuln_only = { + key: metadata + for key, metadata in vuln_meta.items() + if fixed_meta.get(key, {}).get("text") != metadata["text"] + } + fixed_only = { + key: metadata + for key, metadata in fixed_meta.items() + if vuln_meta.get(key, {}).get("text") != metadata["text"] + } + return vuln_only, fixed_only + + def collect_patch_symbols(self): + """ + Collect all changed symbols across every file modified by the + commit, grouped by programming language. + + For each changed file, the analyzer: + - Retrieves the vulnerable and fixed file contents. + - Computes which lines were removed and added. + - Extracts symbols that intersect those changed lines using + Tree-sitter parsing SymbolExtractor + - Diffs the extracted symbols to find those whose body text + actually changed. + - Buckets the results by programming language. + """ + by_language = {} + changed_files = self.get_changed_files() + + for file_path, texts in changed_files.items(): + vulnerable_text = texts["vulnerable_text"] + fixed_text = texts["fixed_text"] + removed_lines, added_lines = self.compute_changed_lines( + vulnerable_text, fixed_text + ) + + vuln_meta, fixed_meta, language = self.analyze( + vulnerable_text=vulnerable_text, + fixed_text=fixed_text, + removed_lines=removed_lines, + added_lines=added_lines, + file_path=file_path, + ) + + if not language or not (vuln_meta or fixed_meta): + continue + + language_bucket = by_language.setdefault( + language, {"vulnerable": {}, "fixed": {}} + ) + + language_bucket["vulnerable"].update( + {f"{file_path}::{key}": metadata for key, metadata in vuln_meta.items()} + ) + language_bucket["fixed"].update( + { + f"{file_path}::{key}": metadata + for key, metadata in fixed_meta.items() + } + ) + + return by_language + + @classmethod + def build_symbol_metadata(cls, nodes, extractor): + """ + Build metadata dictionaries for a list of Tree-sitter AST nodes + representing changed symbols. + + For each node, the qualified name, body text, SHA-256 fingerprint, + and start/end line numbers are extracted and stored in a + dictionary keyed by the qualified name. If duplicate qualified + names are encountered, a numeric suffix is appended to disambiguate. + """ + if not nodes or not extractor: + return {} + + index = extractor.extract_definitions_index() + + metadata = {} + name_counts = {} + for node in nodes: + qualified_name = extractor._build_qualified_name(node, index) + if not qualified_name: + continue + + body_text = node.text.decode("utf-8", errors="replace") + fingerprints = create_sha256_fingerprint(body_text) + + count = name_counts[qualified_name] = name_counts.get(qualified_name, 0) + 1 + key = qualified_name if count == 1 else f"{qualified_name}#{count}" + + metadata[key] = { + "qualified_name": qualified_name, + "text": body_text, + "fingerprint": fingerprints, + "start_line": node.start_point[0] + 1, + "end_line": node.end_point[0] + 1, + "node_type": node.type, + } + return metadata + + @classmethod + def analyze( + cls, vulnerable_text, fixed_text, removed_lines, added_lines, file_path + ): + """ + Analyze the vulnerable and fixed versions of a single file to + extract changed symbols. + + The method performs the following steps: + + - Detects the programming language of the file (using the + fixed version first, falling back to the vulnerable version). + - Verifies the language is supported by Tree-sitter queries. + - Parses both versions into ASTs using Tree-sitter. + - Extracts symbols whose line ranges intersect the removed + lines (vulnerable side) or added lines (fixed side). + - Builds metadata for each set of changed symbols. + - Diffs the two metadata sets to find symbols whose body text + actually changed between versions. + + """ + vulnerable_text = normalize_text(vulnerable_text) + fixed_text = normalize_text(fixed_text) + + language = detect_language_with_scancode( + file_path, fixed_text + ) or detect_language_with_scancode(file_path, vulnerable_text) + + if not is_supported_language(language): + return {}, {}, language + + lang_query = TS_QUERIES[language]() + + vuln_tree, _ = ( + lang_query.parse_code_to_ast(code_text=vulnerable_text) + if vulnerable_text + else (None, None) + ) + fixed_tree, _ = ( + lang_query.parse_code_to_ast(code_text=fixed_text) + if fixed_text + else (None, None) + ) + + if vuln_tree is None and fixed_tree is None: + return {}, {}, language + + vuln_meta_all = {} + fixed_meta_all = {} + + if vuln_tree: + vuln_extractor = SymbolExtractor( + lang_query=lang_query, root_node=vuln_tree.root_node + ) + vuln_nodes = vuln_extractor.extract_changed_symbols( + changed_lines=removed_lines + ) + vuln_meta_all = cls.build_symbol_metadata( + nodes=vuln_nodes, extractor=vuln_extractor + ) + + if fixed_tree: + fixed_extractor = SymbolExtractor( + lang_query=lang_query, root_node=fixed_tree.root_node + ) + fixed_nodes = fixed_extractor.extract_changed_symbols( + changed_lines=added_lines + ) + fixed_meta_all = cls.build_symbol_metadata( + nodes=fixed_nodes, extractor=fixed_extractor + ) + + vuln_meta, fixed_meta = cls.diff_changed_symbols( + vuln_meta=vuln_meta_all, fixed_meta=fixed_meta_all + ) + return vuln_meta, fixed_meta, language + + +def classify_reachability(tool_details): + """ + Classify the reachability status of a vulnerability based on the + collected tool_details from ResourcePatchMatcher. + """ + if not tool_details: + return ReachabilityStatus.NOT_REACHABLE + + status = ReachabilityStatus.NOT_REACHABLE + 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("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 + + if (is_imported or is_defined) and not is_exact: + status = ReachabilityStatus.UNKNOWN + + return status + + +class ResourceAnalyzer: + def __init__(self, resource_text, language): + self.resource_text = normalize_text(resource_text) + self.language = language + + def process_node( + self, node, extractor, definitions_index, definitions, fingerprints + ): + """ + Process a single AST node to extract its qualified name, add it + to the definitions set, compute its fingerprint, and add the + fingerprint to the fingerprints set. + """ + qualified_name = extractor._build_qualified_name(node, definitions_index) + if not qualified_name: + return None + + definitions.add(qualified_name) + body_text = node.text.decode("utf-8", errors="replace") + fingerprint = create_sha256_fingerprint(body_text) + + if fingerprint: + fingerprints.add(fingerprint) + + return qualified_name + + def build_index(self): + """ + Build the full symbol index for the resource by parsing it + with Tree-sitter and extracting all definitions, fingerprints, + imports, and the reverse call graph. + + The method iterates over all functions, classes, and constants + in the resource's AST. For functions, it also extracts call + expressions to populate the callers_of reverse call graph. + """ + if not is_supported_language(self.language) or not self.resource_text: + return None + + lang_query = TS_QUERIES[self.language]() + tree, _ = lang_query.parse_code_to_ast(self.resource_text) + + if tree is None: + return None + + extractor = SymbolExtractor(lang_query=lang_query, root_node=tree.root_node) + definitions_index = extractor.extract_definitions_index() + imports_map = extractor.extract_imports() + separator = extractor.syntax_config.get("separator", ".") + + definitions = set() + fingerprints = set() + callers_of = {} # callee_name -> set of caller_qualified_names + + for node, _ in lang_query.get_functions(tree.root_node): + self.process_node( + node, extractor, definitions_index, definitions, fingerprints + ) + + for node, _ in lang_query.get_classes(tree.root_node): + self.process_node( + node, extractor, definitions_index, definitions, fingerprints + ) + + for node, _ in lang_query.get_constants(tree.root_node): + self.process_node( + node, extractor, definitions_index, definitions, fingerprints + ) + + for receiver_node, callee_node in lang_query.get_calls(tree.root_node): + callee_name = callee_node.text.decode("utf-8", errors="replace") + if not callee_name: + continue + + caller_name = None + curr = callee_node + while curr is not None: + def_info = definitions_index.get(curr.id) + if def_info and def_info.get("qualified_name"): + caller_name = def_info["qualified_name"] + break + curr = curr.parent + + callers_of.setdefault(callee_name, set()).add(caller_name) + + return { + "definitions": definitions, + "fingerprints": fingerprints, + "imports": imports_map, + "callers_of": callers_of, + "separator": separator, + } + + +class ResourcePatchMatcher: + def __init__(self, resource_index): + self.resource_index = resource_index + self.definitions = resource_index.get("definitions", set()) + self.fingerprints = resource_index.get("fingerprints", set()) + 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 tool_details for each matched symbol. + """ + if not patch_symbols_metadata or not self.resource_index: + return {} + + matched = {} + for metadata in patch_symbols_metadata.values(): + qualified_name = metadata["qualified_name"] + fingerprint = metadata["fingerprint"] + defined = qualified_name in self.definitions + 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) + + possible_call_names = {qualified_name} + if imported or defined: + possible_call_names.add(short_name) + possible_call_names.update(import_call_names) + + 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, + "is_called": False, + "is_defined": False, + "is_imported": False, + "is_exact": False, + "reachable_from": [], + }, + ) + + 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["reachable_from"] = sorted([c for c in callers if c is not None]) + + return matched + + +def save_resource_reachability_report(resource, commit_hash, vcs_url, new_report): + """ + Save a reachability report for commit_hash and vcs_url. + If duplicates exist, replace the old report with the new one. + """ + cleaned_reports = [] + replaced = False + + old_reports = resource.extra_data.get("symbols_reachability", []) + for old_report in old_reports: + patch_info = old_report.get("patch", {}) + old_commit_hash = patch_info.get("commit_hash") + old_vcs_url = patch_info.get("vcs_url") + + if old_commit_hash == commit_hash and old_vcs_url == vcs_url: + if not replaced: + cleaned_reports.append(new_report) + replaced = True + # Skip old duplicate + else: + cleaned_reports.append(old_report) + + if not replaced: + cleaned_reports.append(new_report) + + resource.update_extra_data({"symbols_reachability": cleaned_reports}) + + +def get_vulnerabilities_patches(package_vulnerabilities, dependency_vulnerabilities): + """Get unique patch for all vulnerabilities.""" + patches = {} + for vulnerability in package_vulnerabilities + dependency_vulnerabilities: + advisory_uid = vulnerability.get("advisory_uid") + for patch in vulnerability.get("fixed_in_patches", []): + vcs_url = patch.get("vcs_url") + commit_hash = patch.get("commit_hash") + + p_key = (vcs_url, commit_hash) + if p_key not in patches: + patches[p_key] = { + "vcs_url": vcs_url, + "commit_hash": commit_hash, + "advisory_uids": [], + } + + if advisory_uid and advisory_uid not in patches[p_key]["advisory_uids"]: + patches[p_key]["advisory_uids"].append(advisory_uid) + + return list(patches.values()) + + +def collect_resource_index(candidate_resources, logger=None): + """Collect resources symbols for each resource""" + resource_indexes = {} + resources_count = len(candidate_resources) + progress = LoopProgress(resources_count, logger) + for resource in progress.iter(candidate_resources): + resource_language = resource.programming_language + + file_content = normalize_text(resource.file_content) + if not file_content: + continue + + resource_analyzer = ResourceAnalyzer( + resource_text=file_content, language=resource_language + ) + + resource_index = resource_analyzer.build_index() + if resource_index: + resource_indexes[resource.path] = resource_index + + return resource_indexes + + +def collect_patch_symbols(patches, logger=None): + """ + For each unique repo clone it once, + collect patch symbols for all related commits + """ + patch_symbols = {} + patches_by_repo = {} + for patch in patches: + vcs_url = patch.get("vcs_url") + patches_by_repo.setdefault(vcs_url, []).append(patch) + + repo_count = len(patches_by_repo) + repo_progress = LoopProgress(repo_count, logger) + for vcs_url, repo_patches in repo_progress.iter(patches_by_repo.items()): + with tempfile.TemporaryDirectory(prefix="symbol-reachability-") as repo_path: + try: + repo = Repo.clone_from(vcs_url, repo_path) + except Exception as e: + raise Exception(f"Failed to clone repository {vcs_url}: {e!r}") + + try: + for patch in repo_patches: + commit_hash = patch.get("commit_hash") + patch_analyzer = PatchAnalyzer(repo=repo, commit_hash=commit_hash) + patch_symbols[commit_hash] = patch_analyzer.collect_patch_symbols() + except Exception as e: + raise Exception( + f"Failed to collect patch symbols " + f"for {vcs_url}, patch: {repo_patches}: {e!r}" + ) + + return patch_symbols + + +def match_patches_to_resources( + patches, patch_symbols, candidate_resources, resource_indexes, logger=None +): + """Match resource symbols against patch symbols.""" + patches_count = len(patches) + patch_progress = LoopProgress(patches_count, logger=logger) + for patch in patch_progress.iter(patches): + vcs_url = patch.get("vcs_url") + commit_hash = patch.get("commit_hash") + advisory_uids = patch.get("advisory_uids", []) + + patch_symbols_by_language = patch_symbols.get(commit_hash, {}) + if not patch_symbols_by_language: + continue + + for resource in candidate_resources: + resource_index = resource_indexes.get(resource.path) + if not resource_index: + continue + + lang_patch_symbols = patch_symbols_by_language.get( + resource.programming_language + ) + if not lang_patch_symbols: + continue + + vulnerable_symbols = lang_patch_symbols.get("vulnerable", {}) + fixed_symbols = lang_patch_symbols.get("fixed", {}) + + matcher = ResourcePatchMatcher(resource_index=resource_index) + vuln_details = matcher.match(vulnerable_symbols) + fixed_details = matcher.match(fixed_symbols) + + if not any([vuln_details, fixed_details]): + continue + + report = { + "patch": { + "vcs_url": vcs_url, + "commit_hash": commit_hash, + }, + "advisory_uids": advisory_uids, + "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, + } + + save_resource_reachability_report( + resource=resource, + commit_hash=commit_hash, + vcs_url=vcs_url, + new_report=report, + ) + + +def generate_advisory_reachability_report(project, patches, candidate_resources): + """ + 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: "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. + Otherwise, it is NOT_REACHABLE. + """ + status_priority = { + ReachabilityStatus.REACHABLE.value: 3, + ReachabilityStatus.UNKNOWN.value: 2, + ReachabilityStatus.NOT_REACHABLE.value: 1, + } + + advisories_reachability_report = { + "purl": project.purl, + "advisories": [], + } + + advisory_map = {} + for patch in patches: + for adv_uid in patch.get("advisory_uids", []): + 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 + advisories_reachability_report["advisories"].append(adv_data) + + for resource in candidate_resources: + for report in resource.extra_data.get("symbols_reachability", []): + advisory_uids = report.get("advisory_uids", []) + is_reachable = ( + report.get("is_reachable") or ReachabilityStatus.NOT_REACHABLE.value + ) + patch = report.get("patch", {}) + + for adv_uid in advisory_uids: + 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 + advisories_reachability_report["advisories"].append(adv_data) + + tool_details = { + "resource_path": resource.path, + "patch": patch, + "is_reachable": is_reachable, + "tool_details": report.get("tool_details", []), + "vulnerable_symbols": report.get("vulnerable_symbols", []), + "fixed_symbols": report.get("fixed_symbols", []), + } + + advisory_map[adv_uid]["details"].append(tool_details) + + 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 = project.get_output_file_path("reachability", "json") + + with open(reachability_output_path, "w") as f: + json.dump(advisories_reachability_report, f, indent=2) + + return advisories_reachability_report + + +def inject_reachability_data(vulns, advisory_map): + """ + Inject reachability data into a list of vulnerabilities. + Returns True if any vulnerability was updated, False otherwise. + """ + updated = False + for vuln in vulns: + adv_uid = vuln.get("advisory_uid") + if adv_uid in advisory_map: + adv_data = advisory_map[adv_uid] + vuln["is_reachable"] = adv_data.get("is_reachable", "unknown") + vuln["reachability_analysis"] = adv_data.get("details", []) + updated = True + + return updated + + +def apply_reachability_to_packages_and_dependencies(project, advisory_report): + """ + Update DiscoveredPackage and DiscoveredDependency records by injecting the + computed reachability data into their affected_by_vulnerabilities JSON field. + """ + advisories = advisory_report.get("advisories", []) + if not advisories: + return + + advisory_map = {adv["advisory_uid"]: adv for adv in advisories} + targets = ( + (project.discoveredpackages.all(), DiscoveredPackage), + (project.discovereddependencies.all(), DiscoveredDependency), + ) + + for queryset, model in targets: + unsaved = [ + item + for item in queryset + if inject_reachability_data( + item.affected_by_vulnerabilities or [], advisory_map + ) + ] + if unsaved: + model.objects.bulk_update( + objs=unsaved, + fields=["affected_by_vulnerabilities"], + batch_size=10, + ) diff --git a/scanpipe/pipes/symbols.py b/scanpipe/pipes/symbols.py index 76493d8dac..3bd97694f2 100644 --- a/scanpipe/pipes/symbols.py +++ b/scanpipe/pipes/symbols.py @@ -20,6 +20,11 @@ # 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 hashlib +import importlib +from abc import ABC +from functools import cache + from django.db.models import Q from aboutcode.pipeline import LoopProgress @@ -171,3 +176,353 @@ def _collect_and_store_tree_sitter_symbols_and_strings(resource): "source_strings": result.get("source_strings"), } ) + + +@cache +def load_language(language): + from source_inspector import symbols_tree_sitter + + if language not in symbols_tree_sitter.TS_LANGUAGE_WHEELS: + raise ValueError(f"Unsupported language: {language}") + + wheel = symbols_tree_sitter.TS_LANGUAGE_WHEELS[language]["wheel"] + try: + grammar = importlib.import_module(wheel) + except ModuleNotFoundError as exc: + raise symbols_tree_sitter.TreeSitterWheelNotInstalled( + f"Grammar wheel '{wheel}' is not installed." + ) from exc + return symbols_tree_sitter.Language(grammar.language()) + + +def create_sha256_fingerprint(text): + if not text: + return None + + text = text.encode("utf-8", errors="replace") + return hashlib.sha256(text).hexdigest() + + +class LanguageQuery(ABC): + language_name: str = "" + constants_query: str = "" + functions_query: str = "" + classes_query: str = "" + calls_query: str = "" + imports_query: str = "" + syntax_config: dict = { + "self_keyword": None, + "separator": ".", + "wildcard_symbol": None, + } + + def __init__(self): + from tree_sitter import Query + + self.ts_language = load_language(self.language_name) + self._compiled_queries = {} + + for kind in ("constants", "functions", "classes", "calls", "imports"): + source = getattr(self, f"{kind}_query", "").strip() + self._compiled_queries[kind] = ( + Query(self.ts_language, source) if source else None + ) + + def parse_code_to_ast(self, code_text): + from source_inspector import symbols_tree_sitter + from tree_sitter import Parser + + if not code_text: + return None, None + parser = Parser(language=self.ts_language) + return parser.parse( + code_text.encode("utf-8") + ), symbols_tree_sitter.TS_LANGUAGE_WHEELS[self.language_name] + + def run_query(self, kind, root_node): + query = self._compiled_queries.get(kind) + return query.matches(root_node) if query else [] + + def get_functions(self, root_node): + for _, captures in self.run_query("functions", root_node): + def_nodes = captures.get("function") + if not def_nodes: + continue + name_nodes = captures.get("name") + name = ( + name_nodes[0].text.decode("utf-8", errors="replace") + if name_nodes + else None + ) + yield def_nodes[0], name + + def get_classes(self, root_node): + for _, captures in self.run_query("classes", root_node): + def_nodes = captures.get("class") + if not def_nodes: + continue + name_nodes = captures.get("name") + name = ( + name_nodes[0].text.decode("utf-8", errors="replace") + if name_nodes + else None + ) + yield def_nodes[0], name + + def get_calls(self, node): + """Yield raw (receiver_node, callee_node).""" + seen_callees = set() + for _, captures in self.run_query("calls", node): + for callee_node in captures.get("callee", []): + if callee_node.id in seen_callees: + continue + seen_callees.add(callee_node.id) + + receiver_nodes = captures.get("receiver") + receiver_node = receiver_nodes[0] if receiver_nodes else None + yield receiver_node, callee_node + + def get_imports(self, root_node): + """Yield raw (module_name, [(import_name, alias), ...]).""" + for _, captures in self.run_query("imports", root_node): + flat_captures = [] + for tag, nodes in captures.items(): + for node in nodes: + text = node.text.decode("utf-8", errors="replace").strip("'\"") + flat_captures.append((node.start_byte, tag, text)) + + module_name, current_import = None, None + pairs = [] + for _, tag, text in flat_captures: + if tag == "module_name": + module_name = text + elif tag == "import_name": + if current_import is not None: + pairs.append((current_import, None)) + current_import = text + elif tag == "alias": + pairs.append((current_import, text)) + current_import = None + + if current_import is not None: + pairs.append((current_import, None)) + + yield module_name, pairs + + def get_constants(self, root_node): + """Yield raw (constant_node, name).""" + for _, captures in self.run_query("constants", root_node): + def_nodes = captures.get("constant") + if not def_nodes: + continue + name_nodes = captures.get("name") + name = ( + name_nodes[0].text.decode("utf-8", errors="replace") + if name_nodes + else None + ) + yield def_nodes[0], name + + +class PythonTreeSitterQuery(LanguageQuery): + language_name = "Python" + constants_query = "(assignment left: (identifier) @name) @constant" + functions_query = "(function_definition name: (identifier) @name) @function" + classes_query = "(class_definition name: (identifier) @name) @class" + calls_query = """ + (call function: (identifier) @callee) + (call function: (attribute object: (_) @receiver + attribute: (identifier) @callee)) + """ + imports_query = """ + (import_statement name: (dotted_name) @import_name) + (import_statement name: (aliased_import + name: (dotted_name) @import_name + alias: (identifier) @alias)) + (import_from_statement + module_name: [(dotted_name) (relative_import)] @module_name + name: [ + (dotted_name) @import_name + (aliased_import name: (dotted_name) @import_name + alias: (identifier) @alias) + ]) + (import_from_statement + module_name: [(dotted_name) (relative_import)] @module_name + (wildcard_import) @import_name) + """ + syntax_config = {"self_keyword": "self", "separator": ".", "wildcard_symbol": "*"} + + +class JavaTreeSitterQuery(LanguageQuery): + language_name = "Java" + constants_query = ( + "(field_declaration declarator: " + "(variable_declarator name: (identifier) @name)) @constant" + ) + functions_query = """ + [(method_declaration name: (identifier) @name) + (constructor_declaration name: (identifier) @name)] @function + """ + classes_query = """ + [(class_declaration name: (identifier) @name) + (interface_declaration name: (identifier) @name) + (record_declaration name: (identifier) @name) + (enum_declaration name: (identifier) @name)] @class + """ + calls_query = """ + (method_invocation name: (identifier) @callee) + (method_invocation object: (_) @receiver name: (identifier) @callee) + """ + imports_query = """ + (import_declaration (scoped_identifier) @import_name) + (import_declaration (scoped_identifier) @module_name (asterisk) @import_name) + """ + syntax_config = { + "self_keyword": "this", + "separator": ".", + "wildcard_symbol": "*", + "import_local_name": "last", + } + + +TS_QUERIES = { + "Python": PythonTreeSitterQuery, + "Java": JavaTreeSitterQuery, +} + + +class SymbolExtractor: + def __init__(self, lang_query: LanguageQuery, root_node): + self.lang_query = lang_query + self.root_node = root_node + self.syntax_config = lang_query.syntax_config + + def _build_qualified_name(self, node, index): + """ + Build fully qualified names internally + (e.g., ClassName.function_name or ClassName::function_name). + """ + parts = [] + curr = node + while curr is not None: + definition = index.get(curr.id) + if definition is not None and definition["name"]: + parts.append(definition["name"]) + curr = curr.parent + + separator = self.syntax_config.get("separator", ".") + return separator.join(reversed(parts)) + + def extract_definitions_index(self): + """Build the index of definitions with fully qualified names.""" + index = {} + + for node, name in self.lang_query.get_functions(self.root_node): + index[node.id] = {"node": node, "name": name, "kind": "functions"} + + for node, name in self.lang_query.get_classes(self.root_node): + index[node.id] = {"node": node, "name": name, "kind": "classes"} + + for node, name in self.lang_query.get_constants(self.root_node): + index[node.id] = {"node": node, "name": name, "kind": "constants"} + + for def_info in index.values(): + def_info["qualified_name"] = self._build_qualified_name( + def_info["node"], index + ) + + return index + + def extract_changed_symbols(self, changed_lines): + """Map changed line numbers to their enclosing symbol nodes.""" + if self.root_node is None or not changed_lines: + return [] + + definition_ids = set(self.extract_definitions_index().keys()) + if not definition_ids: + return [] + + seen = set() + enclosing = [] + + for line in changed_lines: + row = max(0, line - 1) + node = self.root_node.descendant_for_point_range((row, 0), (row, 0)) + + while node is not None: + if node.id in definition_ids and node.id not in seen: + seen.add(node.id) + enclosing.append(node) + break + node = node.parent + + return enclosing + + def extract_calls(self, node): + """Extract direct calls into (receiver_name, callee_name) text format.""" + calls = [] + for receiver_node, callee_node in self.lang_query.get_calls(node): + receiver_name = ( + receiver_node.text.decode("utf-8", errors="replace") + if receiver_node + else None + ) + callee_name = callee_node.text.decode("utf-8", errors="replace") + + if callee_name: + calls.append((receiver_name, callee_name)) + + 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.""" + wildcard_sym = self.syntax_config.get("wildcard_symbol") + + import_map = {} + wildcard_modules = [] + + for module_name, pairs in self.lang_query.get_imports(self.root_node): + for imp_name, alias in pairs: + if not imp_name: + continue + + if wildcard_sym is not None and imp_name == wildcard_sym: + if module_name: + wildcard_modules.append(module_name) + continue + + 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 + + if wildcard_modules: + import_map["*"] = wildcard_modules + + return import_map + + +def is_supported_language(language): + """Return True if the language is supported by tree-sitter queries.""" + return bool(language) and language in TS_QUERIES diff --git a/scanpipe/pipes/vulnerablecode.py b/scanpipe/pipes/vulnerablecode.py index 20f5e69b5f..4deaefc704 100644 --- a/scanpipe/pipes/vulnerablecode.py +++ b/scanpipe/pipes/vulnerablecode.py @@ -118,6 +118,7 @@ def bulk_search_by_purl( data = { "purls": purls, "details": True, + "reachability": True, } logger.debug(f"VulnerableCode: url={url} purls_count={len(purls)}") diff --git a/scanpipe/tests/data/reachability/java/app.java b/scanpipe/tests/data/reachability/java/app.java new file mode 100644 index 0000000000..0e0b52b178 --- /dev/null +++ b/scanpipe/tests/data/reachability/java/app.java @@ -0,0 +1,50 @@ +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.Map; + +public class App { + public static boolean debug = true; + + public static class ReportGenerator { + private String baseDir; + + public ReportGenerator(String baseDir) { + this.baseDir = baseDir; + } + + public String getBaseDir() { + return baseDir; + } + } + + public static String serveReport(Map requestPayload) { + ReportGenerator generator = new ReportGenerator("/var/reports"); + String requestedFile = requestPayload.get("file"); + + if (requestedFile == null || requestedFile.isEmpty()) { + return "Error: No file specified"; + } + + // VULNERABLE: Direct concatenation allows Path Traversal + // An attacker passing "../../etc/passwd" could read system files. + String targetPath = buildFilePath(generator, requestedFile); + + try { + if (Files.exists(Paths.get(targetPath))) { + return "Serving content of " + targetPath; + } + } catch (Exception e) { + return "Error: Invalid path"; + } + + return "Error: File not found"; + } + + private static String buildFilePath(ReportGenerator generator, String filename) { + return Paths.get(generator.getBaseDir(), filename).toString(); + } + + public static String unrelatedTopLevelFunction() { + return "I am just here to add AST complexity."; + } +} \ No newline at end of file diff --git a/scanpipe/tests/data/reachability/java/fixed-app.java b/scanpipe/tests/data/reachability/java/fixed-app.java new file mode 100644 index 0000000000..39234131b0 --- /dev/null +++ b/scanpipe/tests/data/reachability/java/fixed-app.java @@ -0,0 +1,60 @@ +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Map; + +public class App { + public static boolean debug = false; + + public static class ReportGenerator { + public String baseDir; + + public ReportGenerator(String baseDir) { + this.baseDir = baseDir; + } + } + + public static String serveReport(Map requestPayload) { + ReportGenerator generator = new ReportGenerator("/var/reports"); + String requestedFile = requestPayload.get("file"); + + if (requestedFile == null || requestedFile.isEmpty()) { + return "Error: No file specified"; + } + + String targetPath; + try { + targetPath = buildFilePath(generator, requestedFile); + } catch (Exception e) { + return "Error: Invalid path"; + } + + try { + if (Files.exists(Paths.get(targetPath))) { + return "Serving content of " + targetPath; + } + } catch (Exception e) { + return "Error: Invalid path"; + } + + return "Error: File not found"; + } + + /** + * FIXED: Validate that the resolved path stays within the base_dir + */ + private static String buildFilePath(ReportGenerator generator, String filename) throws Exception { + Path base = Paths.get(generator.baseDir).toAbsolutePath().normalize(); + Path target = Paths.get(generator.baseDir, filename).toAbsolutePath().normalize(); + + if (!target.startsWith(base)) { + throw new Exception("Path Traversal Detected"); + } + return target.toString(); + } + + + public static String unrelatedTopLevelFunction() { + return "I am just here to add AST complexity."; + } +} \ No newline at end of file diff --git a/scanpipe/tests/data/reachability/java/vuln-app.java b/scanpipe/tests/data/reachability/java/vuln-app.java new file mode 100644 index 0000000000..548ffd9eab --- /dev/null +++ b/scanpipe/tests/data/reachability/java/vuln-app.java @@ -0,0 +1,46 @@ +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.Map; + +public class App { + public static boolean debug = true; + + public static class ReportGenerator { + public String baseDir; + + public ReportGenerator(String baseDir) { + this.baseDir = baseDir; + } + } + + public static String serveReport(Map requestPayload) { + ReportGenerator generator = new ReportGenerator("/var/reports"); + String requestedFile = requestPayload.get("file"); + + if (requestedFile == null || requestedFile.isEmpty()) { + return "Error: No file specified"; + } + + // VULNERABLE: Direct concatenation allows Path Traversal + // An attacker passing "../../etc/passwd" could read system files. + String targetPath = buildFilePath(generator, requestedFile); + + try { + if (Files.exists(Paths.get(targetPath))) { + return "Serving content of " + targetPath; + } + } catch (Exception e) { + return "Error: Invalid path"; + } + + return "Error: File not found"; + } + + private static String buildFilePath(ReportGenerator generator, String filename) { + return Paths.get(generator.baseDir, filename).toString(); + } + + public static String unrelatedTopLevelFunction() { + return "I am just here to add AST complexity."; + } +} \ No newline at end of file diff --git a/scanpipe/tests/data/reachability/python/app.py b/scanpipe/tests/data/reachability/python/app.py new file mode 100644 index 0000000000..dae9bce750 --- /dev/null +++ b/scanpipe/tests/data/reachability/python/app.py @@ -0,0 +1,41 @@ +import os + +debug = False + + +class ReportGenerator: + """A dummy class to test AST class method parsing.""" + + def __init__(self, base_dir): + self.base_dir = base_dir + + +def serve_report(request_payload): + """Top-level function handling a request.""" + generator = ReportGenerator("/var/reports") + requested_file = request_payload.get("file") + + # Helper function nested inside serve_report + def build_file_path(filename): + # VULNERABLE: Direct concatenation allows Path Traversal + # An attacker passing "../../etc/passwd" could read system files. + return os.path.join(generator.base_dir, filename) + + if not requested_file: + return "Error: No file specified" + + target_path = build_file_path(requested_file) + + if os.path.exists(target_path): + return f"Serving content of {target_path}" + + return "Error: File not found" + + +def handle_request(req): + return serve_report(req) + + +def unrelated_top_level_function(): + """Test AST node boundaries.""" + return "I am just here to add AST complexity." diff --git a/scanpipe/tests/data/reachability/python/fixed-app.py b/scanpipe/tests/data/reachability/python/fixed-app.py new file mode 100644 index 0000000000..23f5633e80 --- /dev/null +++ b/scanpipe/tests/data/reachability/python/fixed-app.py @@ -0,0 +1,47 @@ +import os + +debug = True + + +class ReportGenerator: + """A dummy class to test AST class method parsing.""" + + def __init__(self, base_dir): + self.base_dir = base_dir + + +def serve_report(request_payload): + """Top-level function handling a request.""" + generator = ReportGenerator("/var/reports") + requested_file = request_payload.get("file") + + # Helper function nested inside serve_report + def build_file_path(filename): + # FIXED: Validate that the resolved path stays within the base_dir + base = os.path.abspath(generator.base_dir) + target = os.path.abspath(os.path.join(base, filename)) + if not target.startswith(base): + raise ValueError("Path Traversal Detected") + return target + + if not requested_file: + return "Error: No file specified" + + try: + target_path = build_file_path(requested_file) + except ValueError: + return "Error: Invalid path" + + if os.path.exists(target_path): + return f"Serving content of {target_path}" + + return "Error: File not found" + + +def handle_request(req): + return serve_report(req) + + +def unrelated_top_level_function(): + """Test AST node boundaries.""" + return "I am just here to add AST complexity." diff --git a/scanpipe/tests/data/reachability/python/vuln-app.py b/scanpipe/tests/data/reachability/python/vuln-app.py new file mode 100644 index 0000000000..dae9bce750 --- /dev/null +++ b/scanpipe/tests/data/reachability/python/vuln-app.py @@ -0,0 +1,41 @@ +import os + +debug = False + + +class ReportGenerator: + """A dummy class to test AST class method parsing.""" + + def __init__(self, base_dir): + self.base_dir = base_dir + + +def serve_report(request_payload): + """Top-level function handling a request.""" + generator = ReportGenerator("/var/reports") + requested_file = request_payload.get("file") + + # Helper function nested inside serve_report + def build_file_path(filename): + # VULNERABLE: Direct concatenation allows Path Traversal + # An attacker passing "../../etc/passwd" could read system files. + return os.path.join(generator.base_dir, filename) + + if not requested_file: + return "Error: No file specified" + + target_path = build_file_path(requested_file) + + if os.path.exists(target_path): + return f"Serving content of {target_path}" + + return "Error: File not found" + + +def handle_request(req): + return serve_report(req) + + +def unrelated_top_level_function(): + """Test AST node boundaries.""" + return "I am just here to add AST complexity." diff --git a/scanpipe/tests/pipes/test_symbols_reachability.py b/scanpipe/tests/pipes/test_symbols_reachability.py new file mode 100644 index 0000000000..bc9b8a866d --- /dev/null +++ b/scanpipe/tests/pipes/test_symbols_reachability.py @@ -0,0 +1,1413 @@ +# 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 json +import os +import shutil +import sys +import tempfile +from pathlib import Path +from unittest import skipIf +from unittest.mock import MagicMock +from unittest.mock import PropertyMock +from unittest.mock import patch + +from django.test import TestCase + +from scanpipe.models import Project +from scanpipe.pipes import collect_and_create_codebase_resources +from scanpipe.pipes.reachability import PatchAnalyzer +from scanpipe.pipes.reachability import ReachabilityStatus +from scanpipe.pipes.reachability import ResourceAnalyzer +from scanpipe.pipes.reachability import ResourcePatchMatcher +from scanpipe.pipes.reachability import classify_reachability +from scanpipe.pipes.symbols import TS_QUERIES +from scanpipe.pipes.symbols import SymbolExtractor + + +@skipIf(sys.platform == "darwin", "Not supported on macOS") +class SymbolReachabilityPipesTest(TestCase): + data = Path(__file__).parent.parent / "data" / "reachability" + + def setUp(self): + self.project1 = Project.objects.create(name="Analysis") + self.project1.codebase_path.mkdir(parents=True, exist_ok=True) + + @patch.object(Project, "package_vulnerabilities", new_callable=PropertyMock) + def test_end_to_end_symbol_reachability_pipeline( + self, mock_package_vulnerabilities + ): + """ + Global end-to-end test for the symbol reachability pipeline. + Sets up a local git repository with a vulnerable and fixed commit, + then runs the full pipeline against a local codebase resource. + """ + from git import Repo + + vcs_dir = tempfile.mkdtemp(prefix="test-vcs-") + try: + repo = Repo.init(vcs_dir) + with repo.config_writer() as config: + config.set_value("user", "email", "test@example.com") + config.set_value("user", "name", "test") + + file_path = "app.py" + vuln_code = ( + "def process_data(data):\n" + " # Vulnerable logic\n" + " return eval(data)\n" + ) + fixed_code = ( + "def process_data(data):\n # Fixed logic\n return int(data)\n" + ) + + vuln_file = Path(vcs_dir) / file_path + vuln_file.write_text(vuln_code) + repo.index.add([file_path]) + repo.index.commit("Vulnerable commit") + + vuln_file.write_text(fixed_code) + repo.index.add([file_path]) + fixed_commit = repo.index.commit("Fixed commit") + + resource_file = self.project1.codebase_path / "local_app.py" + resource_file.write_text(vuln_code) + collect_and_create_codebase_resources(self.project1) + + resource = self.project1.codebaseresources.get(path="local_app.py") + resource.programming_language = "Python" + resource.save() + + repo_url = Path(vcs_dir).as_uri() + + mock_package_vulnerabilities.return_value = [ + { + "fixed_in_patches": [ + { + "vcs_url": repo_url, + "commit_hash": fixed_commit.hexsha, + } + ] + } + ] + + run = self.project1.add_pipeline("analyze_symbols_reachability") + pipeline = run.make_pipeline_instance() + pipeline.execute() + + resource.refresh_from_db() + results = resource.extra_data.get("symbols_reachability") + + expected_results = [ + { + "patch": { + "vcs_url": repo_url, + "commit_hash": fixed_commit.hexsha, + }, + "is_reachable": ReachabilityStatus.REACHABLE.value, + "tool_details": [ + { + "is_exact": True, + "is_called": False, + "is_defined": True, + "is_imported": False, + "symbol_name": "process_data", + "reachable_from": [], + } + ], + "advisory_uids": [], + "fixed_symbols": ["process_data"], + "vulnerable_symbols": ["process_data"], + } + ] + + self.assertEqual(results, expected_results) + finally: + shutil.rmtree(vcs_dir, ignore_errors=True) + + def test_generate_advisory_reachability_report(self): + """Test the generation of the advisory reachability report""" + run = self.project1.add_pipeline("analyze_symbols_reachability") + pipeline = run.make_pipeline_instance() + + pipeline.patches = [ + {"advisory_uids": ["AVID-1"]}, + {"advisory_uids": ["AVID-2"]}, + ] + pipeline.project.purl = "pkg:pypi/test" + + with tempfile.TemporaryDirectory() as tmpdir: + output_file = os.path.join(tmpdir, "reachability.json") + pipeline.project.get_output_file_path = MagicMock(return_value=output_file) + + res1 = MagicMock() + res1.path = "src/file1.py" + res1.extra_data = { + "symbols_reachability": [ + { + "advisory_uids": ["AVID-1", "AVID-2"], + "reachability_status": ReachabilityStatus.REACHABLE.value, + "patch": { + "vcs_url": "https://example.com", + "commit_hash": "abc123", + }, + "evidence": [ + { + "symbol_name": "vuln_sym1", + "called": True, + "defined": False, + "imported": False, + "fingerprint": "88ad9e67c53aa5f7c4" + "3ec4aa52ed34b7930068c9", + "reachable_from": [], + } + ], + "vulnerable_symbols": ["vuln_sym1"], + "fixed_symbols": ["fixed_sym1"], + } + ] + } + + res2 = MagicMock() + res2.path = "src/file2.py" + res2.extra_data = { + "symbols_reachability": [ + { + "advisory_uids": ["AVID-1"], + "reachability_status": ReachabilityStatus.UNKNOWN.value, + "patch": { + "vcs_url": "https://example.com", + "commit_hash": "def456", + }, + "evidence": [], + "vulnerable_symbols": [], + "fixed_symbols": [], + } + ] + } + + res3 = MagicMock() + res3.path = "src/file3.py" + res3.extra_data = { + "symbols_reachability": [ + { + "advisory_uids": ["AVID-2"], + "reachability_status": ReachabilityStatus.NOT_REACHABLE.value, + "patch": { + "vcs_url": "https://example2.com", + "commit_hash": "46a4asf", + }, + "evidence": [], + "vulnerable_symbols": [], + "fixed_symbols": [], + } + ] + } + + res_empty = MagicMock() + res_empty.path = "src/empty.py" + res_empty.extra_data = {} + + pipeline.candidate_resources = [res2, res3, res_empty, res1] + pipeline.generate_advisory_reachability_report() + + with open(output_file) as f: + report = json.load(f) + + expected_report = { + "purl": "pkg:pypi/test", + "advisories": [ + { + "advisory_uid": "AVID-1", + "is_reachable": ReachabilityStatus.NOT_REACHABLE.value, + "details": [ + { + "resource_path": "src/file2.py", + "patch": { + "vcs_url": "https://example.com", + "commit_hash": "def456", + }, + "is_reachable": ReachabilityStatus.NOT_REACHABLE.value, + "tool_details": [], + "vulnerable_symbols": [], + "fixed_symbols": [], + }, + { + "resource_path": "src/file1.py", + "patch": { + "vcs_url": "https://example.com", + "commit_hash": "abc123", + }, + "is_reachable": ReachabilityStatus.NOT_REACHABLE.value, + "tool_details": [], + "vulnerable_symbols": ["vuln_sym1"], + "fixed_symbols": ["fixed_sym1"], + }, + ], + }, + { + "advisory_uid": "AVID-2", + "is_reachable": "no", + "details": [ + { + "resource_path": "src/file3.py", + "patch": { + "vcs_url": "https://example2.com", + "commit_hash": "46a4asf", + }, + "is_reachable": ReachabilityStatus.NOT_REACHABLE.value, + "tool_details": [], + "vulnerable_symbols": [], + "fixed_symbols": [], + }, + { + "resource_path": "src/file1.py", + "patch": { + "vcs_url": "https://example.com", + "commit_hash": "abc123", + }, + "is_reachable": ReachabilityStatus.NOT_REACHABLE.value, + "tool_details": [], + "vulnerable_symbols": ["vuln_sym1"], + "fixed_symbols": ["fixed_sym1"], + }, + ], + }, + ], + } + + self.assertEqual(report, expected_report) + + def _run_reachability_pipeline( + self, + mock_package_vulnerabilities, + mock_collect_symbols, + mock_repo, + file_path, + app_text, + vuln_text, + fixed_text, + expected_results, + ): + """Shared helper to run the end-to-end reachability pipeline.""" + analyzer = PatchAnalyzer(repo=MagicMock(), commit_hash="dummy") + + removed_lines, added_lines = analyzer.compute_changed_lines( + vulnerable_text=vuln_text, fixed_text=fixed_text + ) + vuln_meta, fixed_meta, lang = analyzer.analyze( + vulnerable_text=vuln_text, + fixed_text=fixed_text, + removed_lines=removed_lines, + added_lines=added_lines, + file_path=file_path, + ) + + self.assertTrue(lang) + self.assertTrue(vuln_meta or fixed_meta) + + mock_package_vulnerabilities.return_value = [ + { + "fixed_in_patches": [ + { + "vcs_url": "https://github.com/aboutcode-org/test", + "commit_hash": "07ec0de1964b14bf085a1c9a27ece2b61ab6105c", + } + ] + } + ] + + mock_collect_symbols.return_value = { + lang: { + "vulnerable": { + f"{file_path}::{key}": metadata + for key, metadata in vuln_meta.items() + }, + "fixed": { + f"{file_path}::{key}": metadata + for key, metadata in fixed_meta.items() + }, + } + } + + resource_file = self.project1.codebase_path / file_path + resource_file.parent.mkdir(parents=True, exist_ok=True) + resource_file.write_text(app_text) + collect_and_create_codebase_resources(self.project1) + + resource = self.project1.codebaseresources.get(path=file_path) + resource.programming_language = lang + resource.save() + + run = self.project1.add_pipeline("analyze_symbols_reachability") + pipeline = run.make_pipeline_instance() + pipeline.execute() + + resource.refresh_from_db() + results = resource.extra_data.get("symbols_reachability") + self.assertEqual(results, expected_results) + + @patch("scanpipe.pipes.reachability.Repo") + @patch("scanpipe.pipes.reachability.PatchAnalyzer.collect_patch_symbols") + @patch.object(Project, "package_vulnerabilities", new_callable=PropertyMock) + def test_python_get_symbol_reachability_results( + self, + mock_package_vulnerabilities, + mock_collect_symbols, + mock_repo, + ): + """Test the end-to-end reachability pipeline for Python.""" + file_path = "app.py" + app_text = (self.data / "python" / file_path).read_text() + vuln_text = (self.data / "python" / "vuln-app.py").read_text() + fixed_text = (self.data / "python" / "fixed-app.py").read_text() + + expected_results = [ + { + "patch": { + "vcs_url": "https://github.com/aboutcode-org/test", + "commit_hash": "07ec0de1964b14bf085a1c9a27ece2b61ab6105c", + }, + "is_reachable": "yes", + "tool_details": [ + { + "is_exact": True, + "is_called": False, + "is_defined": True, + "is_imported": False, + "symbol_name": "debug", + "reachable_from": [], + }, + { + "is_exact": True, + "is_called": True, + "is_defined": True, + "is_imported": False, + "symbol_name": "serve_report.build_file_path", + "reachable_from": ["serve_report.target_path"], + }, + { + "is_exact": True, + "is_called": True, + "is_defined": True, + "is_imported": False, + "symbol_name": "serve_report", + "reachable_from": ["handle_request"], + }, + ], + "advisory_uids": [], + "fixed_symbols": [ + "debug", + "serve_report", + "serve_report.build_file_path", + ], + "vulnerable_symbols": [ + "debug", + "serve_report", + "serve_report.build_file_path", + ], + } + ] + + self._run_reachability_pipeline( + mock_package_vulnerabilities, + mock_collect_symbols, + mock_repo, + file_path, + app_text, + vuln_text, + fixed_text, + expected_results, + ) + + @patch("scanpipe.pipes.reachability.Repo") + @patch("scanpipe.pipes.reachability.PatchAnalyzer.collect_patch_symbols") + @patch.object(Project, "package_vulnerabilities", new_callable=PropertyMock) + def test_dependency_simple_reachability( + self, + mock_package_vulnerabilities, + mock_collect_symbols, + mock_repo, + ): + """ + Test the reachability pipeline + for a vulnerability dependency. + """ + file_path = "app.py" + + app_text = ( + "from aiohttp.http_parser import HttpRequestParser\n\n" + "HttpRequestParser.parse_message()\n" + ) + + vuln_text = ( + "class HttpRequestParser:\n" + " def parse_message(self):\n" + " return eval('1')\n" + ) + + fixed_text = ( + "class HttpRequestParser:\n" + " def parse_message(self):\n" + " return int('1')\n" + ) + + expected_results = [ + { + "patch": { + "vcs_url": "https://github.com/aboutcode-org/test", + "commit_hash": "07ec0de1964b14bf085a1c9a27ece2b61ab6105c", + }, + "is_reachable": "yes", + "tool_details": [ + { + "is_exact": False, + "is_called": True, + "is_defined": False, + "is_imported": True, + "symbol_name": "HttpRequestParser.parse_message", + "reachable_from": [], + } + ], + "advisory_uids": [], + "fixed_symbols": ["HttpRequestParser.parse_message"], + "vulnerable_symbols": ["HttpRequestParser.parse_message"], + } + ] + + self._run_reachability_pipeline( + mock_package_vulnerabilities, + mock_collect_symbols, + mock_repo, + file_path, + app_text, + vuln_text, + fixed_text, + expected_results, + ) + + @patch("scanpipe.pipes.reachability.Repo") + @patch("scanpipe.pipes.reachability.PatchAnalyzer.collect_patch_symbols") + @patch.object(Project, "package_vulnerabilities", new_callable=PropertyMock) + def test_java_get_symbol_reachability_results( + self, + mock_package_vulnerabilities, + mock_collect_symbols, + mock_repo, + ): + """Test the end-to-end reachability pipeline for Java.""" + file_path = "app.java" + app_text = (self.data / "java" / file_path).read_text() + vuln_text = (self.data / "java" / "vuln-app.java").read_text() + fixed_text = (self.data / "java" / "fixed-app.java").read_text() + + expected_results = [ + { + "patch": { + "vcs_url": "https://github.com/aboutcode-org/test", + "commit_hash": "07ec0de1964b14bf085a1c9a27ece2b61ab6105c", + }, + "is_reachable": "yes", + "tool_details": [ + { + "is_exact": False, + "is_called": False, + "is_defined": True, + "is_imported": False, + "symbol_name": "App", + "reachable_from": [], + }, + { + "is_exact": True, + "is_called": False, + "is_defined": True, + "is_imported": False, + "symbol_name": "App.serveReport", + "reachable_from": [], + }, + { + "is_exact": False, + "is_called": True, + "is_defined": True, + "is_imported": False, + "symbol_name": "App.buildFilePath", + "reachable_from": ["App.serveReport"], + }, + ], + "advisory_uids": [], + "fixed_symbols": ["App", "App.buildFilePath", "App.serveReport"], + "vulnerable_symbols": ["App", "App.buildFilePath", "App.serveReport"], + } + ] + + self._run_reachability_pipeline( + mock_package_vulnerabilities, + mock_collect_symbols, + mock_repo, + file_path, + app_text, + vuln_text, + fixed_text, + expected_results, + ) + + def test_extract_definitions(self): + """ + Test extracting functions, classes, and constant + definitions from Python code. + """ + source_code = """ +price = 0 +class OrderManager: + def __init__(self, order_id): + self.order_id = order_id + + def process_payment(self): + print("Processing...") + +def calculate_discount(price): + return price * 0.10 + +class InventoryItem: + pass +""" + + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast(code_text=source_code) + functions = list(lang_query.get_functions(tree.root_node)) + + self.assertEqual( + len(functions), 3 + ) # '__init__', 'process_payment', and 'calculate_discount' + + self.assertEqual(functions[0][0].type, "function_definition") + first_func_text = functions[0][0].text.decode("utf-8") + self.assertIn("def __init__", first_func_text) + + classes = list(lang_query.get_classes(tree.root_node)) + self.assertEqual(len(classes), 2) + second_class_text = classes[1][0].text.decode("utf-8") + self.assertIn("class InventoryItem", second_class_text) + + constants = list(lang_query.get_constants(tree.root_node)) + self.assertEqual(len(constants), 1) + + def test_extract_definitions_empty(self): + """Test parsing empty or None source code""" + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast("") + + self.assertIsNone(tree) + + tree_none, _ = lang_query.parse_code_to_ast(None) + self.assertIsNone(tree_none) + + def test_get_qualified_name_functions(self): + """Test building qualified names for nested and top-level functions.""" + source_code = """ +class CoreService: + class Validator: + def validate_payload(self, data): + return True + +def global_utility(): + pass + """ + + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast(source_code) + + extractor = SymbolExtractor(lang_query=lang_query, root_node=tree.root_node) + index = extractor.extract_definitions_index() + + functions = list(lang_query.get_functions(tree.root_node)) + self.assertEqual(len(functions), 2) + + outer_function_name = extractor._build_qualified_name(functions[0][0], index) + inner_function_name = extractor._build_qualified_name(functions[1][0], index) + + self.assertEqual(outer_function_name, "CoreService.Validator.validate_payload") + self.assertEqual(inner_function_name, "global_utility") + + def test_get_qualified_classes(self): + """Test building qualified names for nested class definitions.""" + source_code = """ +class FleetManagement: + class DroneController: + pass + """ + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast(source_code) + + extractor = SymbolExtractor(lang_query=lang_query, root_node=tree.root_node) + index = extractor.extract_definitions_index() + + classes = list(lang_query.get_classes(tree.root_node)) + self.assertEqual(len(classes), 2) + + outer_class_name = extractor._build_qualified_name(classes[0][0], index) + inner_class_name = extractor._build_qualified_name(classes[1][0], index) + + self.assertEqual(outer_class_name, "FleetManagement") + self.assertEqual(inner_class_name, "FleetManagement.DroneController") + + def test_classify_reachability(self): + """ + Test reachability classification for fingerprint + and evidence-based results. + """ + self.assertEqual(classify_reachability(None), ReachabilityStatus.NOT_REACHABLE) + self.assertEqual(classify_reachability({}), ReachabilityStatus.NOT_REACHABLE) + self.assertEqual( + classify_reachability({"tool_details": {}}), + ReachabilityStatus.NOT_REACHABLE, + ) + self.assertEqual( + classify_reachability({"tool_details": {"is_exact": True}}), + ReachabilityStatus.REACHABLE, + ) + + self.assertEqual( + classify_reachability( + {"tool_details": {"is_imported": True, "is_called": True}} + ), + ReachabilityStatus.REACHABLE, + ) + self.assertEqual( + classify_reachability( + {"tool_details": {"is_imported": True, "is_called": False}} + ), + ReachabilityStatus.UNKNOWN, + ) + self.assertEqual( + classify_reachability( + {"tool_details": {"is_imported": False, "is_called": False}} + ), + ReachabilityStatus.NOT_REACHABLE, + ) + + def test_build_symbol_metadata_processing(self): + """Test building metadata for nested changed symbols with deduplication.""" + source_code = """ +class Controller: + def process_data(payload): + def inner_helper(): + return True + return payload.strip() + +if True: + def process_data(payload): + return payload +""" + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast(source_code) + extractor = SymbolExtractor(lang_query=lang_query, root_node=tree.root_node) + vuln_nodes = extractor.extract_changed_symbols( + changed_lines=[1, 2, 3, 4, 5, 6, 7, 8, 9] + ) + metadata = PatchAnalyzer.build_symbol_metadata( + nodes=vuln_nodes, extractor=extractor + ) + self.assertEqual( + metadata, + { + "Controller": { + "qualified_name": "Controller", + "text": "class Controller:\n" + " def process_data(payload):\n" + " def inner_helper():\n " + " return True\n " + " return payload.strip()", + "fingerprint": "de81abd637e27302d8e19c41eab8f4" + "fb6b8abdd9fc4f1fb31d354bc7b23f6d4d", + "start_line": 2, + "end_line": 6, + "node_type": "class_definition", + }, + "Controller.process_data": { + "qualified_name": "Controller.process_data", + "text": "def process_data(payload):\n" + " def inner_helper():\n" + " return True\n" + " return payload.strip()", + "fingerprint": "b0d0ad9a92209a6d79b84e932ce3" + "02a8bc9054a405131adf7dc21e06e2e7c0c1", + "start_line": 3, + "end_line": 6, + "node_type": "function_definition", + }, + "Controller.process_data.inner_helper": { + "qualified_name": "Controller.process_data.inner_helper", + "text": "def inner_helper():\n return True", + "fingerprint": "ee2e246e01e960826cb39a9466e58095" + "d209fdd1cbf8458630be430b3371d6a3", + "start_line": 4, + "end_line": 5, + "node_type": "function_definition", + }, + }, + ) + + def test_diff_changed_symbols(self): + """Test that changed, added, and removed symbols are correctly identified.""" + vuln_meta = { + "serve_report": { + "qualified_name": "app.serve_report", + "text": "def serve_report():\n return os.path.join(base, filename)", + }, + "sanitize_input": { + "qualified_name": "app.sanitize_input", + "text": "def sanitize_input(x):\n return x.strip()", + }, + "deprecated_logger": { + "qualified_name": "app.deprecated_logger", + "text": "def deprecated_logger():\n print('legacy')", + }, + } + + fixed_meta = { + "serve_report": { + "qualified_name": "app.serve_report", + "text": "def serve_report():\n " + " if not target.startswith(base): " + "raise ValueError\n " + " return target", + }, + "sanitize_input": { + "qualified_name": "app.sanitize_input", + "text": "def sanitize_input(x):\n return x.strip()", + }, + "audit_trail": { + "qualified_name": "app.audit_trail", + "text": "def audit_trail():\n log.info('action')", + }, + } + + vuln_only, fixed_only = PatchAnalyzer.diff_changed_symbols( + vuln_meta, fixed_meta + ) + self.assertEqual( + vuln_only, + { + "serve_report": { + "qualified_name": "app.serve_report", + "text": "def serve_report():\n " + " return os.path.join(base, filename)", + }, + "deprecated_logger": { + "qualified_name": "app.deprecated_logger", + "text": "def deprecated_logger():\n print('legacy')", + }, + }, + ) + self.assertEqual( + fixed_only, + { + "serve_report": { + "qualified_name": "app.serve_report", + "text": "def serve_report():\n if not target.startswith(base): " + "raise ValueError\n return target", + }, + "audit_trail": { + "qualified_name": "app.audit_trail", + "text": "def audit_trail():\n log.info('action')", + }, + }, + ) + + def test_analyze_patched_file(self): + """Test analyzing a patched file and extracting changed symbol metadata.""" + vuln_text = (self.data / "python" / "vuln-app.py").read_text(encoding="utf-8") + fixed_text = (self.data / "python" / "fixed-app.py").read_text(encoding="utf-8") + file_path = "python/app.py" + removed_lines, added_lines = PatchAnalyzer.compute_changed_lines( + vuln_text, fixed_text + ) + + vuln_meta, fixed_meta, lang = PatchAnalyzer.analyze( + vulnerable_text=vuln_text, + fixed_text=fixed_text, + removed_lines=removed_lines, + added_lines=added_lines, + file_path=file_path, + ) + + self.assertEqual( + vuln_meta, + { + "debug": { + "qualified_name": "debug", + "text": "debug = False", + "fingerprint": "336908735214468b103dbde11c3ff" + "bd2f76ac9212b8514f831cfa078a67892df", + "start_line": 3, + "end_line": 3, + "node_type": "assignment", + }, + "serve_report.build_file_path": { + "qualified_name": "serve_report.build_file_path", + "text": "def build_file_path(filename):\n" + " # VULNERABLE: Direct concatenation " + "allows Path Traversal\n " + ' # An attacker passing "../../etc/passwd" ' + "could read system files.\n" + " return os.path.join(generator.base_dir, filename)", + "fingerprint": "762e4f7d03b1bf4359c3ca364e55814" + "0239913bfabcc5aa77156460c2eb0a355", + "start_line": 19, + "end_line": 22, + "node_type": "function_definition", + }, + "serve_report": { + "qualified_name": "serve_report", + "text": "def serve_report(request_payload):\n " + ' """Top-level function handling a request."""\n' + ' generator = ReportGenerator("/var/reports")\n' + ' requested_file = request_payload.get("file")\n\n' + " # Helper function nested inside serve_report\n" + " def build_file_path(filename):\n " + " # VULNERABLE: Direct " + "concatenation allows Path Traversal\n " + " # An attacker passing " + '"../../etc/passwd" could read system files.\n' + " return os.path.join(generator.base_dir, filename)\n\n" + " if not requested_file:\n " + ' return "Error: No file specified"\n\n ' + " target_path = build_file_path(requested_file)\n\n " + " if os.path.exists(target_path):\n " + ' return f"Serving content of {target_path}"\n\n ' + ' return "Error: File not found"', + "fingerprint": "d7675efb263896da2a3c006795118" + "33553907e7e6ea619115a6dfc8625c3457e", + "start_line": 13, + "end_line": 32, + "node_type": "function_definition", + }, + }, + ) + + self.assertEqual( + fixed_meta, + { + "debug": { + "qualified_name": "debug", + "text": "debug = True", + "fingerprint": "55d2e2010de610fd32f0c28bc49f535" + "3d6ac60afc70adc5713aa4b675646590e", + "start_line": 3, + "end_line": 3, + "node_type": "assignment", + }, + "serve_report.build_file_path": { + "qualified_name": "serve_report.build_file_path", + "text": "def build_file_path(filename):\n " + " # FIXED: Validate that the resolved" + " path stays within the base_dir\n " + " base = os.path.abspath(generator.base_dir)\n " + " target = os.path.abspath(os.path.join(base, filename))\n " + " if not target.startswith(base):\n " + ' raise ValueError("Path Traversal Detected")\n ' + " return target", + "fingerprint": "646743b5d5497f6ea3b96f860bcbe" + "b38096ce008ad16d2b9a9c3f77a98faca80", + "start_line": 19, + "end_line": 25, + "node_type": "function_definition", + }, + "serve_report": { + "qualified_name": "serve_report", + "text": "def serve_report(request_payload):\n " + ' """Top-level function handling a request."""\n ' + ' generator = ReportGenerator("/var/reports")\n ' + ' requested_file = request_payload.get("file")\n\n ' + " # Helper function nested inside serve_report\n " + "def build_file_path(filename):\n " + " # FIXED: Validate that the" + " resolved path stays within the base_dir\n " + " base = os.path.abspath(generator.base_dir)\n " + " target = os.path.abspath(os.path.join(base, filename))\n" + " if not target.startswith(base):\n " + ' raise ValueError("Path Traversal Detected")\n ' + " return target\n\n " + " if not requested_file:\n " + ' return "Error: No file specified"\n\n try:\n ' + " target_path = build_file_path(requested_file)\n" + " except ValueError:\n " + ' return "Error: Invalid path"\n\n ' + " if os.path.exists(target_path):\n " + ' return f"Serving content of {target_path}"\n\n ' + ' return "Error: File not found"', + "fingerprint": "2deedb21d5f9b1409c59f0b1e55" + "12d73d9afdfc3f469ccf86e8835915d240e76", + "start_line": 13, + "end_line": 38, + "node_type": "function_definition", + }, + }, + ) + + def test_extract_symbols(self): + """Test extracting only the nested function containing the changed line.""" + source_code = ( + "def serve_report(request):\n" # Line 1 (Row 0) + " # Some processing here\n" # Line 2 (Row 1) + " def build_path(filename):\n" # Line 3 (Row 2) + " return filename.strip()\n" # Line 4 (Row 3) <- Targeted Change + " return build_path(request)\n" # Line 5 (Row 4) + ) + + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast(source_code) + + changed_lines = [4] + extractor = SymbolExtractor(lang_query=lang_query, root_node=tree.root_node) + changed_symbols = extractor.extract_changed_symbols(changed_lines) + + self.assertEqual(len(changed_symbols), 1) + target_node = changed_symbols[0] + self.assertEqual(target_node.type, "function_definition") + + node_text = target_node.text.decode("utf-8") + self.assertIn("def build_path", node_text) + self.assertNotIn("def serve_report", node_text) + + def test_extract_symbols_deduplication(self): + """ + Test that multiple changed lines within + the same function produce only a single enclosing symbol. + """ + source_code = ( + "def calculate_total(price, tax):\n" + " amount = price * tax\n" # Line 2 -> Changed + " return price + amount\n" # Line 3 -> Changed + ) + + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast(source_code) + + changed_lines = [2, 3] + symbol_extractor = SymbolExtractor( + lang_query=lang_query, root_node=tree.root_node + ) + enclosing_symbols = symbol_extractor.extract_changed_symbols(changed_lines) + self.assertEqual(len(enclosing_symbols), 1) + self.assertEqual(enclosing_symbols[0].type, "function_definition") + + def test_extract_direct(self): + """ + Test that direct function calls are + extracted from the syntax tree. + """ + source_code = """ +def hello(): + return 10 +def clean_function(): + x = 10 + y = 20 + return hello() + x + y + """.strip() + + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast(code_text=source_code) + extractor = SymbolExtractor(lang_query=lang_query, root_node=tree.root_node) + result = extractor.extract_calls(node=tree.root_node) + self.assertEqual( + result, + [(None, "hello")], + ) + + def test_extract_direct_calls(self): + """Test extraction of direct function calls.""" + python_source = """ +self.update() +process_data() +user.save() + """.strip() + + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast(code_text=python_source) + extractor = SymbolExtractor(lang_query=lang_query, root_node=tree.root_node) + python_calls = extractor.extract_calls(node=tree.root_node) + + expected_python = [ + ("self", "update"), + (None, "process_data"), + ("user", "save"), + ] + self.assertEqual(expected_python, python_calls) + + def test_extract_imports(self): + """ + Test extraction of imported symbols and their + fully qualified module paths. + """ + source_code = """ +from django.db import models +import os.path +import numpy as np +from a.b import c as d +from . import utils +from ..core import engine +from math import * + """.strip() + + lang_query = TS_QUERIES["Python"]() + tree, _ = lang_query.parse_code_to_ast(code_text=source_code) + extractor = SymbolExtractor(lang_query=lang_query, root_node=tree.root_node) + result = extractor.extract_imports() + + expected_map = { + "models": "django.db.models", + "os": "os.path", + "np": "numpy", + "d": "a.b.c", + "utils": ".utils", + "engine": "..core.engine", + "*": ["math"], + } + + self.assertEqual(result, expected_map) + + def test_resource_patch_matcher_python(self): + """ + Test matching + Python patch symbols against imports and calls. + """ + vuln_text = """ +def direct_func(): + return eval("1") + +def aliased_func(): + return eval("2") + +def wildcard_func(): + return eval("3") + +def multiline_func(): + return eval("6") + +def deep_func(): + return eval("7") + +def relative_func(): + return eval("8") + +def module_func(): + return eval("9") + +def unused_func(): + return eval("10") + +class MyClass: + def class_method(self): + return eval("4") + + def target_method(self): + return eval("11") + +class InnerClass: + def deep_method(self): + return eval("5") + """.strip() + + fixed_text = """ +def direct_func(): + return int("1") + +def aliased_func(): + return int("2") + +def wildcard_func(): + return int("3") + +def multiline_func(): + return int("6") + +def deep_func(): + return int("7") + +def relative_func(): + return int("8") + +def module_func(): + return int("9") + +def unused_func(): + return int("10") + +class MyClass: + def class_method(self): + return int("4") + + def target_method(self): + return int("11") + + class InnerClass: + def deep_method(self): + return int("5") + """.strip() + + app_text = """ +from my_module import direct_func +from my_module import aliased_func as af +from my_module import MyClass as C +from my_module import wildcard_func +from my_module import * +from a.b import deep_func +from .relative_module import relative_func +from my_module import ( + multiline_func, +) +from my_module import module_func as mf +from my_module import unused_func + +# 1. Module-level call via alias +mf() + +def execute(): + direct_func() + af() + wildcard_func() + multiline_func() + deep_func() + relative_func() + + c = C() + c.class_method() + c.target_method() + + inner = C.InnerClass() + inner.deep_method() + +def execute_nested(): + def inner_helper(): + # 2. Call inside a nested function via alias + mf() + inner_helper() + +def wildcard_caller(): + # 3. Call via wildcard import + module_func() + +# unused_func is imported but never called +""".strip() + + file_path = "my_module.py" + analyzer = PatchAnalyzer(repo=MagicMock(), commit_hash="dummy") + removed_lines, added_lines = analyzer.compute_changed_lines( + vuln_text, fixed_text + ) + + vuln_meta, fixed_meta, lang = analyzer.analyze( + vulnerable_text=vuln_text, + fixed_text=fixed_text, + removed_lines=removed_lines, + added_lines=added_lines, + file_path=file_path, + ) + + patch_symbols_by_language = { + lang: { + "vulnerable": { + f"{file_path}::{key}": metadata + for key, metadata in vuln_meta.items() + }, + "fixed": { + f"{file_path}::{key}": metadata + for key, metadata in fixed_meta.items() + }, + } + } + + resource_analyzer = ResourceAnalyzer(resource_text=app_text, language=lang) + resource_index = resource_analyzer.build_index() + matcher = ResourcePatchMatcher(resource_index) + + vulnerable_symbols = patch_symbols_by_language[lang]["vulnerable"] + fixed_symbols = patch_symbols_by_language[lang]["fixed"] + + vuln_details = matcher.match(vulnerable_symbols) + fixed_details = matcher.match(fixed_symbols) + + matched = {**vuln_details, **fixed_details} + self.assertIn("direct_func", matched) + self.assertTrue(matched["direct_func"]["is_imported"]) + self.assertTrue(matched["direct_func"]["is_called"]) + self.assertEqual(matched["direct_func"]["reachable_from"], ["execute"]) + + self.assertIn("aliased_func", matched) + self.assertTrue(matched["aliased_func"]["is_imported"]) + self.assertTrue(matched["aliased_func"]["is_called"]) + self.assertEqual(matched["aliased_func"]["reachable_from"], ["execute"]) + + self.assertIn("wildcard_func", matched) + self.assertTrue(matched["wildcard_func"]["is_imported"]) + self.assertTrue(matched["wildcard_func"]["is_called"]) + self.assertEqual(matched["wildcard_func"]["reachable_from"], ["execute"]) + + self.assertIn("MyClass.class_method", matched) + self.assertTrue(matched["MyClass.class_method"]["is_imported"]) + self.assertTrue(matched["MyClass.class_method"]["is_called"]) + self.assertEqual(matched["MyClass.class_method"]["reachable_from"], ["execute"]) + + self.assertIn("MyClass.InnerClass.deep_method", matched) + self.assertTrue(matched["MyClass.InnerClass.deep_method"]["is_imported"]) + self.assertTrue(matched["MyClass.InnerClass.deep_method"]["is_called"]) + self.assertEqual( + matched["MyClass.InnerClass.deep_method"]["reachable_from"], ["execute"] + ) + + self.assertIn("multiline_func", matched) + self.assertTrue(matched["multiline_func"]["is_imported"]) + self.assertTrue(matched["multiline_func"]["is_called"]) + self.assertEqual(matched["multiline_func"]["reachable_from"], ["execute"]) + + self.assertIn("deep_func", matched) + self.assertTrue(matched["deep_func"]["is_imported"]) + self.assertTrue(matched["deep_func"]["is_called"]) + self.assertEqual(matched["deep_func"]["reachable_from"], ["execute"]) + + self.assertIn("relative_func", matched) + self.assertTrue(matched["relative_func"]["is_imported"]) + self.assertTrue(matched["relative_func"]["is_called"]) + self.assertEqual(matched["relative_func"]["reachable_from"], ["execute"]) + + self.assertIn("MyClass.target_method", matched) + self.assertTrue(matched["MyClass.target_method"]["is_imported"]) + self.assertTrue(matched["MyClass.target_method"]["is_called"]) + self.assertEqual( + matched["MyClass.target_method"]["reachable_from"], ["execute"] + ) + + self.assertIn("module_func", matched) + self.assertTrue(matched["module_func"]["is_imported"]) + self.assertTrue(matched["module_func"]["is_called"]) + self.assertEqual( + matched["module_func"]["reachable_from"], + ["execute_nested.inner_helper", "wildcard_caller"], + ) + + self.assertIn("unused_func", matched) + self.assertTrue(matched["unused_func"]["is_imported"]) + self.assertFalse(matched["unused_func"]["is_called"]) + self.assertEqual(matched["unused_func"]["reachable_from"], []) + + def test_resource_patch_matcher_java(self): + """ + Test matching + Java patch symbols against imports and calls. + """ + vuln_text = """ +package com.example; + +public class Service { + public void processRequest(String input) { + Runtime.getRuntime().exec(input); + } + + public void utilityMethod() { + Runtime.getRuntime().exec("cmd"); + } +} + """.strip() + + fixed_text = """ +package com.example; + +public class Service { + public void processRequest(String input) { + System.out.println(input); + } + + public void utilityMethod() { + System.out.println("cmd"); + } +} + """.strip() + + app_text = """ +package com.test; + +import com.example.Service; +import com.example.*; + +public class Main { + public void execute(String data) { + Service service = new Service(); + service.processRequest(data); + + service.utilityMethod(); + } +} + """.strip() + + file_path = "Service.java" + analyzer = PatchAnalyzer(repo=MagicMock(), commit_hash="dummy") + removed_lines, added_lines = analyzer.compute_changed_lines( + vuln_text, fixed_text + ) + + vuln_meta, fixed_meta, lang = analyzer.analyze( + vulnerable_text=vuln_text, + fixed_text=fixed_text, + removed_lines=removed_lines, + added_lines=added_lines, + file_path=file_path, + ) + + patch_symbols_by_language = { + lang: { + "vulnerable": { + f"{file_path}::{key}": metadata + for key, metadata in vuln_meta.items() + }, + "fixed": { + f"{file_path}::{key}": metadata + for key, metadata in fixed_meta.items() + }, + } + } + + resource_analyzer = ResourceAnalyzer(resource_text=app_text, language=lang) + resource_index = resource_analyzer.build_index() + matcher = ResourcePatchMatcher(resource_index) + + vulnerable_symbols = patch_symbols_by_language[lang]["vulnerable"] + fixed_symbols = patch_symbols_by_language[lang]["fixed"] + + vuln_details = matcher.match(vulnerable_symbols) + fixed_details = matcher.match(fixed_symbols) + + matched = {**vuln_details, **fixed_details} + self.assertIn("Service.processRequest", matched) + self.assertTrue(matched["Service.processRequest"]["is_imported"]) + self.assertTrue(matched["Service.processRequest"]["is_called"]) + self.assertEqual( + matched["Service.processRequest"]["reachable_from"], ["Main.execute"] + ) + + self.assertIn("Service.utilityMethod", matched) + self.assertTrue(matched["Service.utilityMethod"]["is_imported"]) + self.assertTrue(matched["Service.utilityMethod"]["is_called"]) + self.assertEqual( + matched["Service.utilityMethod"]["reachable_from"], ["Main.execute"] + ) diff --git a/scanpipe/tests/test_api.py b/scanpipe/tests/test_api.py index 3fb58b3f7f..90719208d9 100644 --- a/scanpipe/tests/test_api.py +++ b/scanpipe/tests/test_api.py @@ -56,6 +56,7 @@ from scanpipe.models import WebhookSubscription from scanpipe.pipes.input import copy_input from scanpipe.pipes.output import JSONResultsGenerator +from scanpipe.pipes.reachability import apply_reachability_to_packages_and_dependencies from scanpipe.tests import dependency_data1 from scanpipe.tests import filter_warnings from scanpipe.tests import make_message @@ -1374,3 +1375,85 @@ def test_scanpipe_api_serializer_get_serializer_fields(self): with self.assertRaises(LookupError): get_serializer_fields(None) + + def test_scanpipe_api_project_action_package_with_reachability(self): + self.discovered_package1.affected_by_vulnerabilities = [ + { + "advisory_id": "PYSEC-2026-1", + "advisory_uid": "pypa/scancode/PYSEC-2026-1", + "summary": "summary 1", + "risk_score": 1, + }, + { + "advisory_id": "PYSEC-2026-2", + "advisory_uid": "pypa/scancode/PYSEC-2026-2", + "summary": "summary 2", + "risk_score": 2, + }, + { + "advisory_id": "PYSEC-2026-3", + "advisory_uid": "pypa/scancode/PYSEC-2026-3", + "summary": "summary 3", + "risk_score": 3, + }, + ] + self.discovered_package1.save() + advisory_map = { + "purl": "pkg:pypi/daglib@0.3.2", + "advisories": [ + { + "advisory_uid": "pypa/scancode/PYSEC-2026-1", + "is_reachable": "unknown", + "details": [ + { + "resource_path": "scancode/session.py", + "is_reachable": "unknown", + "vulnerable_symbols": ["SqliteAccountInfo"], + } + ], + }, + { + "advisory_uid": "pypa/scancode/PYSEC-2026-2", + "is_reachable": "yes", + "details": [ + { + "resource_path": "b2sdk/session.py", + "is_reachable": "yes", + "vulnerable_symbols": ["SqliteAccountInfo"], + } + ], + }, + ], + } + + apply_reachability_to_packages_and_dependencies(self.project1, advisory_map) + url = reverse("project-packages", args=[self.project1.uuid]) + response = self.csrf_client.get(url) + + self.assertEqual(status.HTTP_200_OK, response.status_code) + self.assertEqual(1, response.data["count"]) + + pkg_response = response.data["results"][0] + vulns = pkg_response["affected_by_vulnerabilities"] + + self.assertEqual(3, len(vulns)) + + self.assertEqual("pypa/scancode/PYSEC-2026-1", vulns[0]["advisory_uid"]) + self.assertEqual("unknown", vulns[0]["is_reachable"]) + self.assertIn("reachability_analysis", vulns[0]) + self.assertEqual(1, len(vulns[0]["reachability_analysis"])) + self.assertEqual( + "scancode/session.py", vulns[0]["reachability_analysis"][0]["resource_path"] + ) + + self.assertEqual("pypa/scancode/PYSEC-2026-2", vulns[1]["advisory_uid"]) + self.assertEqual("yes", vulns[1]["is_reachable"]) + self.assertIn("reachability_analysis", vulns[1]) + self.assertEqual(1, len(vulns[1]["reachability_analysis"])) + self.assertEqual( + "b2sdk/session.py", vulns[1]["reachability_analysis"][0]["resource_path"] + ) + + self.assertEqual("pypa/scancode/PYSEC-2026-3", vulns[2]["advisory_uid"]) + self.assertNotIn("is_reachable", vulns[2]) + self.assertNotIn("reachability_analysis", vulns[2])