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..2489df9c49 --- /dev/null +++ b/scanpipe/pipelines/analyze_symbols_reachability.py @@ -0,0 +1,33 @@ +# +# Copyright (c) nexB Inc. and others. All rights reserved. +# VulnerableCode is a trademark of nexB Inc. +# SPDX-License-Identifier: Apache-2.0 +# See http://www.apache.org/licenses/LICENSE-2.0 for the license text. +# See https://github.com/aboutcode-org/vulnerablecode for support or download. +# See https://aboutcode.org for more information about nexB OSS projects. +# + +from scanpipe.pipelines import Pipeline +from scanpipe.pipes import reachability + + +class SymbolReachability(Pipeline): + """Patch reachability analysis for given vulnerability patches.""" + + download_inputs = False + is_addon = True + results_url = "/project/{slug}/resources/?extra_data=symbol_reachability" + + @classmethod + def steps(cls): + return (cls.analyze_symbol_reachability,) + + def analyze_symbol_reachability(self): + """ + Perform symbol-level reachability analysis for each patch. This step compares + the AST of patched/vulnerable files against the codebase resources. + Results are stored directly in the 'extra_data' of each CodebaseResource. + """ + reachability.analyze_and_store_symbol_reachability_results( + project=self.project, logger=self.log + ) diff --git a/scanpipe/pipes/reachability.py b/scanpipe/pipes/reachability.py new file mode 100644 index 0000000000..6c7768dbad --- /dev/null +++ b/scanpipe/pipes/reachability.py @@ -0,0 +1,568 @@ +# 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 os +import shutil +import tempfile +from enum import Enum +from pathlib import Path +from typing import Any + +from git import Repo +from git.diff import NULL_TREE +from scancode.api import get_file_info + +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 = "REACHABLE" + POTENTIALLY_REACHABLE = "POTENTIALLY_REACHABLE" + NOT_REACHABLE = "NOT_REACHABLE" + + +def normalize_text(content): + 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): + content = normalize_text(content) + + if not content: + return None + + tmp_dir = tempfile.mkdtemp(prefix="patch-lang-") + + try: + target = Path(tmp_dir) / Path(file_path).name + target.write_text(content, encoding="utf-8", errors="replace") + + info = get_file_info(location=str(target)) or {} + return info.get("programming_language") or None + + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +class GitRepositoryContext: + def __init__(self, vcs_url: str): + self.vcs_url = vcs_url + self.repo_path = None + self._repo = None + + def __enter__(self) -> "GitRepositoryContext": + self.repo_path = tempfile.mkdtemp(prefix="symbol-reachability-") + try: + self._repo = Repo.clone_from(self.vcs_url, self.repo_path) + return self + except Exception as exc: + self._cleanup() + raise ValueError(f"Failed to clone/checkout {self.vcs_url}") from exc + + def __exit__(self, exc_type, exc_val, exc_tb): + self._cleanup() + + @property + def repo(self) -> Repo: + return self._repo + + def _cleanup(self): + if self.repo_path and os.path.exists(self.repo_path): + shutil.rmtree(self.repo_path, ignore_errors=True) + + +class PatchAnalyzer: + def __init__(self, repo: Repo, commit_hash: str): + 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): + 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): + 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): + 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: SymbolExtractor, index: dict = None + ): + if not nodes or not extractor: + return {} + + if index is None: + index = extractor.extract_definitions_index() + + metadata = {} + 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) + + key = qualified_name + suffix = 1 + while key in metadata: + suffix += 1 + key = f"{qualified_name}#{suffix}" + + 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 + ): + 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( + 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 generate_reachability_report(patch, repo, candidate_resources, logger=None): + vcs_url = patch.get("vcs_url") + commit_hash = patch.get("commit_hash") + + try: + patch_analyzer = PatchAnalyzer(repo=repo, commit_hash=commit_hash) + patch_symbols_by_language = patch_analyzer.collect_patch_symbols() + + if not patch_symbols_by_language: + return + + for resource in candidate_resources: + resource_language = resource.programming_language + patch_symbols = patch_symbols_by_language.get(resource_language) + + if not patch_symbols: + continue + + 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 not resource_index: + continue + + matcher = ResourcePatchMatcher(resource_index=resource_index) + vuln_evidence = matcher.match(patch_symbols["vulnerable"]) + fixed_evidence = matcher.match(patch_symbols["fixed"]) + + if not any([vuln_evidence, fixed_evidence]): + continue + + report = { + "symbols_reachability": { + "patch": { + "vcs_url": vcs_url, + "commit_hash": commit_hash, + }, + "evidence": list(vuln_evidence.values()), + "fixed_symbols": sorted(fixed_evidence.keys()), + "vulnerable_symbols": sorted(vuln_evidence.keys()), + "reachability_status": classify_reachability(vuln_evidence).value, + } + } + + existing = resource.extra_data.get("symbols_reachability") + reports = existing if isinstance(existing, list) else [] + + if not any( + r.get("patch", {}).get("commit_hash") == commit_hash for r in reports + ): + reports.append(report) + resource.update_extra_data({"symbols_reachability": reports}) + + except Exception as e: + if logger: + logger( + f"Failed to collect symbol reachability for " + f"{vcs_url}@{commit_hash}: {e}" + ) + + +def analyze_and_store_symbol_reachability_results(project, logger=None): + candidate_resources = project.codebaseresources.files().filter( + is_binary=False, is_archive=False, is_media=False + ) + + for vulnerability in project.package_vulnerabilities: + patches_by_repo = {} + for patch in vulnerability.get("fixed_in_patches", []): + vcs_url = patch.get("vcs_url") + commit_hash = patch.get("commit_hash") + if vcs_url and commit_hash: + patches_by_repo.setdefault(vcs_url, []).append(patch) + + for vcs_url, patches in patches_by_repo.items(): + try: + with GitRepositoryContext(vcs_url) as git_context: + for patch in patches: + commit_hash = patch.get("commit_hash") + git_context.repo.git.checkout(commit_hash) + generate_reachability_report( + patch, git_context.repo, candidate_resources, logger + ) + except Exception as e: + if logger: + logger( + f"Failed to process repository {vcs_url} " + f"for symbol reachability: {e}" + ) + + +def classify_reachability(evidence): + if not evidence: + return ReachabilityStatus.NOT_REACHABLE + + status = ReachabilityStatus.NOT_REACHABLE + for item in evidence.values(): + is_called = bool(item.get("called")) + has_path = bool(item.get("reachable_from")) + is_defined = bool(item.get("defined")) + is_imported = bool(item.get("imported")) + is_exact = bool(item.get("fingerprint")) + + 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.POTENTIALLY_REACHABLE + + return status + + +class ResourceAnalyzer: + def __init__(self, resource_text: str, language: str): + self.resource_text = normalize_text(resource_text) + self.language = language + + def process_node( + self, node, extractor, definitions_index, definitions: set, fingerprints: set + ) -> str | None: + """Extract the qualified name, update definitions, and fingerprint""" + 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) -> dict | None: + 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): + qualified_name = self.process_node( + node, extractor, definitions_index, definitions, fingerprints + ) + if qualified_name: + for _, callee_name in extractor.extract_calls(node): + callers_of.setdefault(callee_name, set()).add(qualified_name) + + 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 + ) + + return { + "definitions": definitions, + "fingerprints": fingerprints, + "imports": imports_map, + "callers_of": callers_of, + "separator": separator, + } + + +class ResourcePatchMatcher: + def __init__(self, resource_index: dict): + 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", ".") + + def match(self, patch_symbols_metadata: dict[str, Any]) -> dict[str, Any]: + 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"] + + short_name = ( + qualified_name.rsplit(self.separator, 1)[-1] + if self.separator in qualified_name + else qualified_name + ) + + defined = qualified_name in self.definitions + fingerprint_hit = bool(fingerprint and fingerprint in self.fingerprints) + + imported = ( + qualified_name in self.imports + or qualified_name in self.imports.values() + ) + + callers = set() + callers.update(self.callers_of.get(short_name, set())) + callers.update(self.callers_of.get(qualified_name, set())) + called = bool(callers) + + if not (defined or fingerprint_hit or called or imported): + continue + + entry = matched.setdefault( + qualified_name, + { + "symbol_name": qualified_name, + "called": False, + "defined": False, + "imported": False, + "fingerprint": None, + "reachable_from": [], + }, + ) + + if defined: + entry["defined"] = True + if imported: + entry["imported"] = True + if called: + entry["called"] = True + entry["reachable_from"] = sorted(callers) + + if fingerprint_hit: + entry["fingerprint"] = fingerprint + + return matched diff --git a/scanpipe/pipes/symbols.py b/scanpipe/pipes/symbols.py index 76493d8dac..f86630b4f5 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,337 @@ def _collect_and_store_tree_sitter_symbols_and_strings(resource): "source_strings": result.get("source_strings"), } ) + + +@cache +def load_language(language: str): + 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: str): + 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: str, 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": "*"} + + +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: dict) -> str: + """ + 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: dict[int, dict] = {} + + 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: list[int]): + """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 extract_imports(self): + """Map every local alias to its absolute imported path.""" + separator = self.syntax_config.get("separator", ".") + wildcard_sym = self.syntax_config.get("wildcard_symbol") + + import_map: dict[str, str | list[str]] = {} + wildcard_modules: list[str] = [] + + 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 = alias or imp_name + if not alias and separator in imp_name: + local_name = imp_name.split(separator)[0] + + if module_name: + if module_name == separator: + absolute_path = f"{separator}{imp_name}" + else: + absolute_path = f"{module_name}{separator}{imp_name}" + else: + absolute_path = imp_name + + 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/app.py b/scanpipe/tests/data/reachability/app.py new file mode 100644 index 0000000000..32eebc2fa1 --- /dev/null +++ b/scanpipe/tests/data/reachability/app.py @@ -0,0 +1,37 @@ +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 unrelated_top_level_function(): + """Test AST node boundaries.""" + return "I am just here to add AST complexity." diff --git a/scanpipe/tests/data/reachability/fixed-app.py b/scanpipe/tests/data/reachability/fixed-app.py new file mode 100644 index 0000000000..30470b6cda --- /dev/null +++ b/scanpipe/tests/data/reachability/fixed-app.py @@ -0,0 +1,43 @@ +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 unrelated_top_level_function(): + """Test AST node boundaries.""" + return "I am just here to add AST complexity." diff --git a/scanpipe/tests/data/reachability/vuln-app.py b/scanpipe/tests/data/reachability/vuln-app.py new file mode 100644 index 0000000000..32eebc2fa1 --- /dev/null +++ b/scanpipe/tests/data/reachability/vuln-app.py @@ -0,0 +1,37 @@ +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 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..0d3ff42fa1 --- /dev/null +++ b/scanpipe/tests/pipes/test_symbols_reachability.py @@ -0,0 +1,679 @@ +# SPDX-License-Identifier: Apache-2.0 +# +# http://nexb.com and https://github.com/nexB/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/nexB/scancode.io for support and download. +import sys +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 analyze_and_store_symbol_reachability_results +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("scanpipe.pipes.reachability.GitRepositoryContext") + @patch("scanpipe.pipes.reachability.PatchAnalyzer.collect_patch_symbols") + @patch.object(Project, "package_vulnerabilities", new_callable=PropertyMock) + def test_get_symbol_reachability_results( + self, + mock_package_vulnerabilities, + mock_collect_symbols, + mock_git_context, + ): + file_path = "app.py" + app_text = (self.data / file_path).read_text() + vuln_text = (self.data / "vuln-app.py").read_text() + fixed_text = (self.data / "fixed-app.py").read_text() + + 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_git_context.return_value.__enter__.return_value.repo = MagicMock() + 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 / "app.py" + resource_file.write_text(app_text) + collect_and_create_codebase_resources(self.project1) + + resource = self.project1.codebaseresources.get(path="app.py") + resource.programming_language = lang + resource.save() + + analyze_and_store_symbol_reachability_results(self.project1) + + resource.refresh_from_db() + results = resource.extra_data.get("symbols_reachability") + + self.assertEqual( + results, + [ + { + "symbols_reachability": { + "patch": { + "vcs_url": "https://github.com/aboutcode-org/test", + "commit_hash": "07ec0de1964b14bf085a1c9a27ece2b61ab6105c", + }, + "evidence": [ + { + "called": False, + "defined": True, + "imported": False, + "fingerprint": "336908735214468b103dbde" + "11c3ffbd2f76ac9212b8514f831cfa078a67892df", + "symbol_name": "debug", + "reachable_from": [], + }, + { + "called": True, + "defined": True, + "imported": False, + "fingerprint": "762e4f7d03b1bf4359c3ca364" + "e558140239913bfabcc5aa77156460c2eb0a355", + "symbol_name": "serve_report.build_file_path", + "reachable_from": ["serve_report"], + }, + { + "called": False, + "defined": True, + "imported": False, + "fingerprint": "d7675efb263896da2a3c0067951183" + "3553907e7e6ea619115a6dfc8625c3457e", + "symbol_name": "serve_report", + "reachable_from": [], + }, + ], + "fixed_symbols": [ + "debug", + "serve_report", + "serve_report.build_file_path", + ], + "vulnerable_symbols": [ + "debug", + "serve_report", + "serve_report.build_file_path", + ], + "reachability_status": "REACHABLE", + } + } + ], + ) + + def test_extract_definitions(self): + 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): + 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): + 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): + 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): + self.assertEqual(classify_reachability(None), ReachabilityStatus.NOT_REACHABLE) + self.assertEqual(classify_reachability({}), ReachabilityStatus.NOT_REACHABLE) + self.assertEqual( + classify_reachability({"evidence": {}}), ReachabilityStatus.NOT_REACHABLE + ) + self.assertEqual( + classify_reachability({"evidence": {"fingerprint": "hash123"}}), + ReachabilityStatus.REACHABLE, + ) + + self.assertEqual( + classify_reachability({"evidence": {"imported": True, "called": True}}), + ReachabilityStatus.REACHABLE, + ) + self.assertEqual( + classify_reachability({"evidence": {"imported": True, "called": False}}), + ReachabilityStatus.POTENTIALLY_REACHABLE, + ) + self.assertEqual( + classify_reachability({"evidence": {"imported": False, "called": False}}), + ReachabilityStatus.NOT_REACHABLE, + ) + + def test_build_symbol_metadata_processing(self): + 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) + index = extractor.extract_definitions_index() + 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, index=index + ) + 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): + 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): + vuln_text = (self.data / "vuln-app.py").read_text(encoding="utf-8") + fixed_text = (self.data / "fixed-app.py").read_text(encoding="utf-8") + file_path = "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): + 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): + 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_collect_imports(self): + 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_extract_direct(self): + 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): + 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): + 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)