From e09966aea19086e77b5fa776cd80b0d43b432203 Mon Sep 17 00:00:00 2001 From: ziad hany Date: Wed, 22 Apr 2026 03:41:41 +0200 Subject: [PATCH 01/13] feat: create pipeline for symbol reachability and add a test Signed-off-by: ziad hany --- .../pipelines/collect_symbols_reachability.py | 35 ++++++++++++++++ scanpipe/tests/data/reachability/app.py | 35 ++++++++++++++++ .../tests/data/reachability/diff-app.patch | 39 ++++++++++++++++++ scanpipe/tests/data/reachability/fixed-app.py | 41 +++++++++++++++++++ scanpipe/tests/data/reachability/vuln-app.py | 35 ++++++++++++++++ 5 files changed, 185 insertions(+) create mode 100644 scanpipe/pipelines/collect_symbols_reachability.py create mode 100644 scanpipe/tests/data/reachability/app.py create mode 100644 scanpipe/tests/data/reachability/diff-app.patch create mode 100644 scanpipe/tests/data/reachability/fixed-app.py create mode 100644 scanpipe/tests/data/reachability/vuln-app.py diff --git a/scanpipe/pipelines/collect_symbols_reachability.py b/scanpipe/pipelines/collect_symbols_reachability.py new file mode 100644 index 0000000000..15519fc661 --- /dev/null +++ b/scanpipe/pipelines/collect_symbols_reachability.py @@ -0,0 +1,35 @@ +# +# 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 a vulnerability patches + """ + + download_inputs = False + is_addon = True + results_url = "/project/{slug}/resources/?extra_data=symbol_reachability" + + @classmethod + def steps(cls): + return (cls.analyze_and_store_symbol_reachability,) + + def analyze_and_store_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.collect_and_store_symbol_reachability_results( + project=self.project, logger=self.log + ) diff --git a/scanpipe/tests/data/reachability/app.py b/scanpipe/tests/data/reachability/app.py new file mode 100644 index 0000000000..c64ae7d9d1 --- /dev/null +++ b/scanpipe/tests/data/reachability/app.py @@ -0,0 +1,35 @@ +import os + + +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(): + """An extra function to test AST node boundaries.""" + return "I am just here to add AST complexity." diff --git a/scanpipe/tests/data/reachability/diff-app.patch b/scanpipe/tests/data/reachability/diff-app.patch new file mode 100644 index 0000000000..ccb86953a8 --- /dev/null +++ b/scanpipe/tests/data/reachability/diff-app.patch @@ -0,0 +1,39 @@ +From 8f7b1c3d9a4e2b6f5d8c1a2e3f4b5c6d7e8f9a0b Mon Sep 17 00:00:00 2001 +From: Security Team +Date: Tue, 2 Jun 2026 10:00:00 +0000 +Subject: [PATCH] Fix path traversal vulnerability in report generator + +- Validates that target paths stay within the designated base_dir. +- Catches ValueError on invalid path resolution. +--- + app.py | 12 +++++++++--- + 1 file changed, 9 insertions(+), 3 deletions(-) + +diff --git a/app.py b/app.py +index a1b2c3d..e4f5g6h 100644 +--- a/app.py ++++ b/app.py +@@ -15,13 +15,19 @@ def serve_report(request_payload): + # 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) ++ # 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" + +- target_path = build_file_path(requested_file) ++ 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}" \ No newline at end of file diff --git a/scanpipe/tests/data/reachability/fixed-app.py b/scanpipe/tests/data/reachability/fixed-app.py new file mode 100644 index 0000000000..3296bb843e --- /dev/null +++ b/scanpipe/tests/data/reachability/fixed-app.py @@ -0,0 +1,41 @@ +import os + + +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(): + """An extra function to 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..c64ae7d9d1 --- /dev/null +++ b/scanpipe/tests/data/reachability/vuln-app.py @@ -0,0 +1,35 @@ +import os + + +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(): + """An extra function to test AST node boundaries.""" + return "I am just here to add AST complexity." From 672194bfa7af9fd4fd13dfb55bb182656cd59dea Mon Sep 17 00:00:00 2001 From: ziad hany Date: Wed, 10 Jun 2026 15:08:45 +0300 Subject: [PATCH 02/13] Fix the format bugs and refactor the code Signed-off-by: ziad hany --- scanpipe/pipelines/collect_symbols_reachability.py | 8 +++----- scanpipe/tests/data/reachability/app.py | 2 +- scanpipe/tests/data/reachability/fixed-app.py | 2 +- scanpipe/tests/data/reachability/vuln-app.py | 2 +- 4 files changed, 6 insertions(+), 8 deletions(-) diff --git a/scanpipe/pipelines/collect_symbols_reachability.py b/scanpipe/pipelines/collect_symbols_reachability.py index 15519fc661..c1d5fb11c4 100644 --- a/scanpipe/pipelines/collect_symbols_reachability.py +++ b/scanpipe/pipelines/collect_symbols_reachability.py @@ -12,9 +12,7 @@ class SymbolReachability(Pipeline): - """ - Patch reachability analysis, for given a vulnerability patches - """ + """Patch reachability analysis for given vulnerability patches.""" download_inputs = False is_addon = True @@ -26,8 +24,8 @@ def steps(cls): def analyze_and_store_symbol_reachability(self): """ - Perform symbol-level reachability analysis for each patch. - This step compares the AST of patched/vulnerable files against the codebase resources. + 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.collect_and_store_symbol_reachability_results( diff --git a/scanpipe/tests/data/reachability/app.py b/scanpipe/tests/data/reachability/app.py index c64ae7d9d1..b8c9eff5e0 100644 --- a/scanpipe/tests/data/reachability/app.py +++ b/scanpipe/tests/data/reachability/app.py @@ -31,5 +31,5 @@ def build_file_path(filename): def unrelated_top_level_function(): - """An extra function to test AST node boundaries.""" + """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 index 3296bb843e..ca5a6f4c8b 100644 --- a/scanpipe/tests/data/reachability/fixed-app.py +++ b/scanpipe/tests/data/reachability/fixed-app.py @@ -37,5 +37,5 @@ def build_file_path(filename): def unrelated_top_level_function(): - """An extra function to test AST node boundaries.""" + """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 index c64ae7d9d1..b8c9eff5e0 100644 --- a/scanpipe/tests/data/reachability/vuln-app.py +++ b/scanpipe/tests/data/reachability/vuln-app.py @@ -31,5 +31,5 @@ def build_file_path(filename): def unrelated_top_level_function(): - """An extra function to test AST node boundaries.""" + """Test AST node boundaries.""" return "I am just here to add AST complexity." From aee4691d42cf25de1b5bf75ff04aa56ccd92e116 Mon Sep 17 00:00:00 2001 From: ziad hany Date: Thu, 16 Jul 2026 02:02:04 +0300 Subject: [PATCH 03/13] Remove dependency and copy only the required file Signed-off-by: ziad hany --- scanpipe/pipes/unidiff/patch.py | 667 ++++++++++++++++++++++++ scanpipe/pipes/unidiff/patch.py.ABOUT | 13 + scanpipe/pipes/unidiff/patch.py.LICENSE | 20 + 3 files changed, 700 insertions(+) create mode 100644 scanpipe/pipes/unidiff/patch.py create mode 100644 scanpipe/pipes/unidiff/patch.py.ABOUT create mode 100644 scanpipe/pipes/unidiff/patch.py.LICENSE diff --git a/scanpipe/pipes/unidiff/patch.py b/scanpipe/pipes/unidiff/patch.py new file mode 100644 index 0000000000..c51bdbef79 --- /dev/null +++ b/scanpipe/pipes/unidiff/patch.py @@ -0,0 +1,667 @@ +# Extracted essential patch code analyzer and modified from the original unidiff library: +# https://github.com/matiasb/python-unidiff/blob/2771a878f7bc6619e625feb4dbad3427f57f5237/unidiff/patch.py + +# -*- coding: utf-8 -*- + +# The MIT License (MIT) +# Copyright (c) 2014-2023 Matias Bordese +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, +# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE +# OR OTHER DEALINGS IN THE SOFTWARE. + +from __future__ import unicode_literals +import re +from io import StringIO +from typing import Iterable, Optional, Union + +class UnidiffParseError(Exception): ... + +open_file = open +make_str = str +implements_to_string = lambda x: x +unicode = str +basestring = str + +RE_SOURCE_FILENAME = re.compile( + r'^--- (?P"?[^\t\n]+"?)(?:\t(?P[^\n]+))?') +RE_TARGET_FILENAME = re.compile( + r'^\+\+\+ (?P"?[^\t\n]+"?)(?:\t(?P[^\n]+))?') + + +# check diff git line for git renamed files support +RE_DIFF_GIT_HEADER = re.compile( + r'^diff --git (?P"?a/[^\t\n]+"?) (?P"?b/[^\t\n]+"?)') +RE_DIFF_GIT_HEADER_URI_LIKE = re.compile( + r'^diff --git (?P.*://[^\t\n]+) (?P.*://[^\t\n]+)') +RE_DIFF_GIT_HEADER_NO_PREFIX = re.compile( + r'^diff --git (?P[^\t\n]+) (?P[^\t\n]+)') + +# check diff git new file marker `deleted file mode 100644` +RE_DIFF_GIT_DELETED_FILE = re.compile(r'^deleted file mode \d+$') + +# check diff git new file marker `new file mode 100644` +RE_DIFF_GIT_NEW_FILE = re.compile(r'^new file mode \d+$') + + +# @@ (source offset, length) (target offset, length) @@ (section header) +RE_HUNK_HEADER = re.compile( + r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))?\ @@[ ]?(.*)") + +# kept line (context) +# \n empty line (treat like context) +# + added line +# - deleted line +# \ No newline case +RE_HUNK_BODY_LINE = re.compile( + r'^(?P[- \+\\])(?P.*)', re.DOTALL) +RE_HUNK_EMPTY_BODY_LINE = re.compile( + r'^(?P[- \+\\]?)(?P[\r\n]{1,2})', re.DOTALL) + +RE_NO_NEWLINE_MARKER = re.compile(r'^\\ No newline at end of file') + +RE_BINARY_DIFF = re.compile( + r'^Binary files? ' + r'(?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?' + r'(?: and (?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?)? (differ|has changed)') + +DEFAULT_ENCODING = 'UTF-8' + +DEV_NULL = '/dev/null' +LINE_TYPE_ADDED = '+' +LINE_TYPE_REMOVED = '-' +LINE_TYPE_CONTEXT = ' ' +LINE_TYPE_EMPTY = '' +LINE_TYPE_NO_NEWLINE = '\\' +LINE_VALUE_NO_NEWLINE = ' No newline at end of file' + +@implements_to_string +class Line(object): + """A diff line.""" + + def __init__(self, value, line_type, + source_line_no=None, target_line_no=None, diff_line_no=None): + # type: (str, str, Optional[int], Optional[int], Optional[int]) -> None + super(Line, self).__init__() + self.source_line_no = source_line_no + self.target_line_no = target_line_no + self.diff_line_no = diff_line_no + self.line_type = line_type + self.value = value + + def __repr__(self): + # type: () -> str + return make_str("") % (self.line_type, self.value) + + def __str__(self): + # type: () -> str + return "%s%s" % (self.line_type, self.value) + + def __eq__(self, other): + # type: (Line) -> bool + return (self.source_line_no == other.source_line_no and + self.target_line_no == other.target_line_no and + self.diff_line_no == other.diff_line_no and + self.line_type == other.line_type and + self.value == other.value) + + @property + def is_added(self): + # type: () -> bool + return self.line_type == LINE_TYPE_ADDED + + @property + def is_removed(self): + # type: () -> bool + return self.line_type == LINE_TYPE_REMOVED + + @property + def is_context(self): + # type: () -> bool + return self.line_type == LINE_TYPE_CONTEXT + + +@implements_to_string +class PatchInfo(list): + """Lines with extended patch info. + + Format of this info is not documented and it very much depends on + patch producer. + + """ + + def __repr__(self): + # type: () -> str + value = "" % self[0].strip() + return make_str(value) + + def __str__(self): + # type: () -> str + return ''.join(unicode(line) for line in self) + + +@implements_to_string +class Hunk(list): + """Each of the modified blocks of a file.""" + + def __init__(self, src_start=0, src_len=0, tgt_start=0, tgt_len=0, + section_header=''): + # type: (int, int, int, int, str) -> None + super(Hunk, self).__init__() + if src_len is None: + src_len = 1 + if tgt_len is None: + tgt_len = 1 + self.source_start = int(src_start) + self.source_length = int(src_len) + self.target_start = int(tgt_start) + self.target_length = int(tgt_len) + self.section_header = section_header + self._added = None # Optional[int] + self._removed = None # Optional[int] + + def __repr__(self): + # type: () -> str + value = "" % (self.source_start, + self.source_length, + self.target_start, + self.target_length, + self.section_header) + return make_str(value) + + def __str__(self): + # type: () -> str + # section header is optional and thus we output it only if it's present + head = "@@ -%d,%d +%d,%d @@%s\n" % ( + self.source_start, self.source_length, + self.target_start, self.target_length, + ' ' + self.section_header if self.section_header else '') + content = ''.join(unicode(line) for line in self) + return head + content + + def append(self, line): + # type: (Line) -> None + """Append the line to hunk, and keep track of source/target lines.""" + # Make sure the line is encoded correctly. This is a no-op except for + # potentially raising a UnicodeDecodeError. + str(line) + super(Hunk, self).append(line) + + @property + def added(self): + # type: () -> Optional[int] + if self._added is not None: + return self._added + # re-calculate each time to allow for hunk modifications + # (which should mean metadata_only switch wasn't used) + return sum(1 for line in self if line.is_added) + + @property + def removed(self): + # type: () -> Optional[int] + if self._removed is not None: + return self._removed + # re-calculate each time to allow for hunk modifications + # (which should mean metadata_only switch wasn't used) + return sum(1 for line in self if line.is_removed) + + def is_valid(self): + # type: () -> bool + """Check hunk header data matches entered lines info.""" + return (len(self.source) == self.source_length and + len(self.target) == self.target_length) + + def source_lines(self): + # type: () -> Iterable[Line] + """Hunk lines from source file (generator).""" + return (l for l in self if l.is_context or l.is_removed) + + @property + def source(self): + # type: () -> Iterable[str] + return [str(l) for l in self.source_lines()] + + def target_lines(self): + # type: () -> Iterable[Line] + """Hunk lines from target file (generator).""" + return (l for l in self if l.is_context or l.is_added) + + @property + def target(self): + # type: () -> Iterable[str] + return [str(l) for l in self.target_lines()] + + +class PatchedFile(list): + """Patch updated file, it is a list of Hunks.""" + + def __init__(self, patch_info=None, source='', target='', + source_timestamp=None, target_timestamp=None, + is_binary_file=False): + # type: (Optional[PatchInfo], str, str, Optional[str], Optional[str], bool, bool) -> None + super(PatchedFile, self).__init__() + self.patch_info = patch_info + self.source_file = source + self.source_timestamp = source_timestamp + self.target_file = target + self.target_timestamp = target_timestamp + self.is_binary_file = is_binary_file + + def __repr__(self): + # type: () -> str + return make_str("") % make_str(self.path) + + def __str__(self): + # type: () -> str + source = '' + target = '' + # patch info is optional + info = '' if self.patch_info is None else str(self.patch_info) + if not self.is_binary_file and self: + source = "--- %s%s\n" % ( + self.source_file, + '\t' + self.source_timestamp if self.source_timestamp else '') + target = "+++ %s%s\n" % ( + self.target_file, + '\t' + self.target_timestamp if self.target_timestamp else '') + hunks = ''.join(unicode(hunk) for hunk in self) + return info + source + target + hunks + + def _parse_hunk(self, header, diff, encoding, metadata_only): + # type: (str, enumerate[str], Optional[str], bool) -> None + """Parse hunk details.""" + header_info = RE_HUNK_HEADER.match(header) + hunk_info = header_info.groups() + hunk = Hunk(*hunk_info) + + source_line_no = hunk.source_start + target_line_no = hunk.target_start + expected_source_end = source_line_no + hunk.source_length + expected_target_end = target_line_no + hunk.target_length + added = 0 + removed = 0 + + for diff_line_no, line in diff: + if encoding is not None: + line = line.decode(encoding) + + if metadata_only: + # quick line type detection, no regex required + line_type = line[0] if line else LINE_TYPE_CONTEXT + if line_type not in (LINE_TYPE_ADDED, + LINE_TYPE_REMOVED, + LINE_TYPE_CONTEXT, + LINE_TYPE_NO_NEWLINE): + raise UnidiffParseError( + 'Hunk diff line expected: %s' % line) + + if line_type == LINE_TYPE_ADDED: + target_line_no += 1 + added += 1 + elif line_type == LINE_TYPE_REMOVED: + source_line_no += 1 + removed += 1 + elif line_type == LINE_TYPE_CONTEXT: + target_line_no += 1 + source_line_no += 1 + + # no file content tracking + original_line = None + + else: + # parse diff line content + valid_line = RE_HUNK_BODY_LINE.match(line) + if not valid_line: + valid_line = RE_HUNK_EMPTY_BODY_LINE.match(line) + + if not valid_line: + raise UnidiffParseError( + 'Hunk diff line expected: %s' % line) + + line_type = valid_line.group('line_type') + if line_type == LINE_TYPE_EMPTY: + line_type = LINE_TYPE_CONTEXT + + value = valid_line.group('value') # type: str + original_line = Line(value, line_type=line_type) + + if line_type == LINE_TYPE_ADDED: + original_line.target_line_no = target_line_no + target_line_no += 1 + elif line_type == LINE_TYPE_REMOVED: + original_line.source_line_no = source_line_no + source_line_no += 1 + elif line_type == LINE_TYPE_CONTEXT: + original_line.target_line_no = target_line_no + original_line.source_line_no = source_line_no + target_line_no += 1 + source_line_no += 1 + elif line_type == LINE_TYPE_NO_NEWLINE: + pass + else: + original_line = None + + # stop parsing if we got past expected number of lines + if (source_line_no > expected_source_end or + target_line_no > expected_target_end): + raise UnidiffParseError('Hunk is longer than expected') + + if original_line: + original_line.diff_line_no = diff_line_no + hunk.append(original_line) + + # if hunk source/target lengths are ok, hunk is complete + if (source_line_no == expected_source_end and + target_line_no == expected_target_end): + break + + # report an error if we haven't got expected number of lines + if (source_line_no < expected_source_end or + target_line_no < expected_target_end): + raise UnidiffParseError('Hunk is shorter than expected') + + if metadata_only: + # HACK: set fixed calculated values when metadata_only is enabled + hunk._added = added + hunk._removed = removed + + self.append(hunk) + + def _add_no_newline_marker_to_last_hunk(self): + # type: () -> None + if not self: + raise UnidiffParseError( + 'Unexpected marker:' + LINE_VALUE_NO_NEWLINE) + last_hunk = self[-1] + last_hunk.append( + Line(LINE_VALUE_NO_NEWLINE + '\n', line_type=LINE_TYPE_NO_NEWLINE)) + + def _append_trailing_empty_line(self): + # type: () -> None + if not self: + raise UnidiffParseError('Unexpected trailing newline character') + last_hunk = self[-1] + last_hunk.append(Line('\n', line_type=LINE_TYPE_EMPTY)) + + @property + def path(self): + # type: () -> str + """Return the file path abstracted from VCS.""" + filepath = self.source_file + if filepath in (None, DEV_NULL) or ( + self.is_rename and self.target_file not in (None, DEV_NULL)): + # if this is a rename, prefer the target filename + filepath = self.target_file + + quoted = filepath.startswith('"') and filepath.endswith('"') + if quoted: + filepath = filepath[1:-1] + + if filepath.startswith('a/') or filepath.startswith('b/'): + filepath = filepath[2:] + + if quoted: + filepath = '"{}"'.format(filepath) + + return filepath + + @property + def added(self): + # type: () -> int + """Return the file total added lines.""" + return sum([hunk.added for hunk in self]) + + @property + def removed(self): + # type: () -> int + """Return the file total removed lines.""" + return sum([hunk.removed for hunk in self]) + + @property + def is_rename(self): + return (self.source_file != DEV_NULL + and self.target_file != DEV_NULL + and self.source_file[2:] != self.target_file[2:]) + + @property + def is_added_file(self): + # type: () -> bool + """Return True if this patch adds the file.""" + if self.source_file == DEV_NULL: + return True + return (len(self) == 1 and self[0].source_start == 0 and + self[0].source_length == 0) + + @property + def is_removed_file(self): + # type: () -> bool + """Return True if this patch removes the file.""" + if self.target_file == DEV_NULL: + return True + return (len(self) == 1 and self[0].target_start == 0 and + self[0].target_length == 0) + + @property + def is_modified_file(self): + # type: () -> bool + """Return True if this patch modifies the file.""" + return not (self.is_added_file or self.is_removed_file) + + +@implements_to_string +class PatchSet(list): + """A list of PatchedFiles.""" + + def __init__(self, f, encoding=None, metadata_only=False): + # type: (Union[StringIO, str], Optional[str], bool) -> None + super(PatchSet, self).__init__() + + # convert string inputs to StringIO objects + if isinstance(f, basestring): + f = self._convert_string(f, encoding) # type: StringIO + + # make sure we pass an iterator object to parse + data = iter(f) + # if encoding is None, assume we are reading unicode data + # when metadata_only is True, only perform a minimal metadata parsing + # (ie. hunks without content) which is around 2.5-6 times faster; + # it will still validate the diff metadata consistency and get counts + self._parse(data, encoding=encoding, metadata_only=metadata_only) + + def __repr__(self): + # type: () -> str + return make_str('') % super(PatchSet, self).__repr__() + + def __str__(self): + # type: () -> str + return ''.join(unicode(patched_file) for patched_file in self) + + def _parse(self, diff, encoding, metadata_only): + # type: (StringIO, Optional[str], bool) -> None + current_file = None + patch_info = None + + diff = enumerate(diff, 1) + for unused_diff_line_no, line in diff: + if encoding is not None: + line = line.decode(encoding) + + # check for a git file rename + is_diff_git_header = RE_DIFF_GIT_HEADER.match(line) or \ + RE_DIFF_GIT_HEADER_URI_LIKE.match(line) or \ + RE_DIFF_GIT_HEADER_NO_PREFIX.match(line) + if is_diff_git_header: + patch_info = PatchInfo() + source_file = is_diff_git_header.group('source') + target_file = is_diff_git_header.group('target') + current_file = PatchedFile( + patch_info, source_file, target_file, None, None) + self.append(current_file) + patch_info.append(line) + continue + + # check for a git new file + is_diff_git_new_file = RE_DIFF_GIT_NEW_FILE.match(line) + if is_diff_git_new_file: + if current_file is None or patch_info is None: + raise UnidiffParseError('Unexpected new file found: %s' % line) + current_file.source_file = DEV_NULL + patch_info.append(line) + continue + + # check for a git deleted file + is_diff_git_deleted_file = RE_DIFF_GIT_DELETED_FILE.match(line) + if is_diff_git_deleted_file: + if current_file is None or patch_info is None: + raise UnidiffParseError('Unexpected deleted file found: %s' % line) + current_file.target_file = DEV_NULL + patch_info.append(line) + continue + + # check for source file header + is_source_filename = RE_SOURCE_FILENAME.match(line) + if is_source_filename: + source_file = is_source_filename.group('filename') + source_timestamp = is_source_filename.group('timestamp') + # reset current file, unless we are processing a rename + # (in that case, source files should match) + if current_file is not None and not ( + current_file.source_file == source_file): + current_file = None + elif current_file is not None: + current_file.source_timestamp = source_timestamp + continue + + # check for target file header + is_target_filename = RE_TARGET_FILENAME.match(line) + if is_target_filename: + target_file = is_target_filename.group('filename') + target_timestamp = is_target_filename.group('timestamp') + if current_file is not None and not (current_file.target_file == target_file): + raise UnidiffParseError('Target without source: %s' % line) + if current_file is None: + # add current file to PatchSet + current_file = PatchedFile( + patch_info, source_file, target_file, + source_timestamp, target_timestamp) + self.append(current_file) + patch_info = None + else: + current_file.target_timestamp = target_timestamp + continue + + # check for hunk header + is_hunk_header = RE_HUNK_HEADER.match(line) + if is_hunk_header: + patch_info = None + if current_file is None: + raise UnidiffParseError('Unexpected hunk found: %s' % line) + current_file._parse_hunk(line, diff, encoding, metadata_only) + continue + + # check for no newline marker + is_no_newline = RE_NO_NEWLINE_MARKER.match(line) + if is_no_newline: + if current_file is None: + raise UnidiffParseError('Unexpected marker: %s' % line) + current_file._add_no_newline_marker_to_last_hunk() + continue + + # sometimes hunks can be followed by empty lines + if line == '\n' and current_file is not None: + current_file._append_trailing_empty_line() + continue + + # if nothing has matched above then this line is a patch info + if patch_info is None: + current_file = None + patch_info = PatchInfo() + + is_binary_diff = RE_BINARY_DIFF.match(line) + if is_binary_diff: + source_file = is_binary_diff.group('source_filename') + target_file = is_binary_diff.group('target_filename') + patch_info.append(line) + if current_file is not None: + current_file.is_binary_file = True + else: + current_file = PatchedFile( + patch_info, source_file, target_file, is_binary_file=True) + self.append(current_file) + patch_info = None + current_file = None + continue + + if line == 'GIT binary patch\n': + current_file.is_binary_file = True + patch_info = None + current_file = None + continue + + patch_info.append(line) + + @classmethod + def from_filename(cls, filename, encoding=DEFAULT_ENCODING, errors=None, newline=None): + # type: (str, str, Optional[str]) -> PatchSet + """Return a PatchSet instance given a diff filename.""" + with open_file(filename, 'r', encoding=encoding, errors=errors, newline=newline) as f: + instance = cls(f) + return instance + + @staticmethod + def _convert_string(data, encoding=None, errors='strict'): + # type: (Union[str, bytes], str, str) -> StringIO + if encoding is not None: + # if encoding is given, assume bytes and decode + data = unicode(data, encoding=encoding, errors=errors) + return StringIO(data) + + @classmethod + def from_string(cls, data, encoding=None, errors='strict'): + # type: (str, str, Optional[str]) -> PatchSet + """Return a PatchSet instance given a diff string.""" + return cls(cls._convert_string(data, encoding, errors)) + + @property + def added_files(self): + # type: () -> list[PatchedFile] + """Return patch added files as a list.""" + return [f for f in self if f.is_added_file] + + @property + def removed_files(self): + # type: () -> list[PatchedFile] + """Return patch removed files as a list.""" + return [f for f in self if f.is_removed_file] + + @property + def modified_files(self): + # type: () -> list[PatchedFile] + """Return patch modified files as a list.""" + return [f for f in self if f.is_modified_file] + + @property + def added(self): + # type: () -> int + """Return the patch total added lines.""" + return sum([f.added for f in self]) + + @property + def removed(self): + # type: () -> int + """Return the patch total removed lines.""" + return sum([f.removed for f in self]) \ No newline at end of file diff --git a/scanpipe/pipes/unidiff/patch.py.ABOUT b/scanpipe/pipes/unidiff/patch.py.ABOUT new file mode 100644 index 0000000000..3118ed9643 --- /dev/null +++ b/scanpipe/pipes/unidiff/patch.py.ABOUT @@ -0,0 +1,13 @@ +about_resource: patch.py constants.py errors.py +name: patch +version: 0.7.5 +download_url: https://github.com/matiasb/python-unidiff/archive/refs/tags/v0.7.5.zip +description: Simple Python library to parse and interact with unified diff data. +homepage_url: https://github.com/matiasb/python-unidiff +license_expression: mit +attribute: yes +package_url: pkg:pypi/unidiff@0.7.5 +licenses: + - key: mit + name: MIT License + file: mit.LICENSE \ No newline at end of file diff --git a/scanpipe/pipes/unidiff/patch.py.LICENSE b/scanpipe/pipes/unidiff/patch.py.LICENSE new file mode 100644 index 0000000000..ca2c04c202 --- /dev/null +++ b/scanpipe/pipes/unidiff/patch.py.LICENSE @@ -0,0 +1,20 @@ +The MIT License (MIT) +Copyright (c) 2012 Matias Bordese + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, +DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE +OR OTHER DEALINGS IN THE SOFTWARE. \ No newline at end of file From 6363f1bbb525d846c8d6988cd8916bb52e6c57b4 Mon Sep 17 00:00:00 2001 From: ziad hany Date: Thu, 23 Jul 2026 02:50:28 +0300 Subject: [PATCH 04/13] Update pipeline/functions name Get the dependency file and included in the PR Signed-off-by: ziad hany --- .../pipelines/collect_symbols_reachability.py | 33 -- scanpipe/pipelines/find_vulnerabilities.py | 12 +- scanpipe/pipes/unidiff/patch.py | 309 +++++++++++------- scanpipe/pipes/vulnerablecode.py | 7 +- 4 files changed, 197 insertions(+), 164 deletions(-) delete mode 100644 scanpipe/pipelines/collect_symbols_reachability.py diff --git a/scanpipe/pipelines/collect_symbols_reachability.py b/scanpipe/pipelines/collect_symbols_reachability.py deleted file mode 100644 index c1d5fb11c4..0000000000 --- a/scanpipe/pipelines/collect_symbols_reachability.py +++ /dev/null @@ -1,33 +0,0 @@ -# -# 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_and_store_symbol_reachability,) - - def analyze_and_store_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.collect_and_store_symbol_reachability_results( - project=self.project, logger=self.log - ) diff --git a/scanpipe/pipelines/find_vulnerabilities.py b/scanpipe/pipelines/find_vulnerabilities.py index b0c8066b9a..1bdf8af6bc 100644 --- a/scanpipe/pipelines/find_vulnerabilities.py +++ b/scanpipe/pipelines/find_vulnerabilities.py @@ -19,7 +19,7 @@ # # 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 aboutcode.pipeline import optional_step from scanpipe.pipelines import Pipeline from scanpipe.pipes import vulnerablecode @@ -34,11 +34,13 @@ class FindVulnerabilities(Pipeline): download_inputs = False is_addon = True results_url = "/project/{slug}/packages/?is_vulnerable=yes" + reachability = False @classmethod def steps(cls): return ( cls.check_vulnerablecode_service_availability, + cls.enable_reachability_analysis, cls.lookup_packages_vulnerabilities, cls.lookup_dependencies_vulnerabilities, ) @@ -48,6 +50,12 @@ def get_availability(cls): if not vulnerablecode.is_configured(): return "VulnerableCode is not configured." + @optional_step("reachability") + def enable_reachability_analysis(self): + """Enable the reachability flag for vulnerability lookups.""" + self.reachability = True + self.log("Reachability analysis is ENABLED.") + def check_vulnerablecode_service_availability(self): """Check if the VulnerableCode service if configured and available.""" if not vulnerablecode.is_configured(): @@ -62,6 +70,7 @@ def lookup_packages_vulnerabilities(self): vulnerablecode.fetch_vulnerabilities( packages=packages, ignore_set=self.project.ignored_vulnerabilities_set, + reachability=self.reachability, logger=self.log, ) @@ -71,5 +80,6 @@ def lookup_dependencies_vulnerabilities(self): vulnerablecode.fetch_vulnerabilities( packages=dependencies, ignore_set=self.project.ignored_vulnerabilities_set, + reachability=self.reachability, logger=self.log, ) diff --git a/scanpipe/pipes/unidiff/patch.py b/scanpipe/pipes/unidiff/patch.py index c51bdbef79..8075a8cc77 100644 --- a/scanpipe/pipes/unidiff/patch.py +++ b/scanpipe/pipes/unidiff/patch.py @@ -1,4 +1,5 @@ -# Extracted essential patch code analyzer and modified from the original unidiff library: +# Extracted essential patch code analyzer and modified +# from the original unidiff library: # https://github.com/matiasb/python-unidiff/blob/2771a878f7bc6619e625feb4dbad3427f57f5237/unidiff/patch.py # -*- coding: utf-8 -*- @@ -24,13 +25,13 @@ # OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE # OR OTHER DEALINGS IN THE SOFTWARE. -from __future__ import unicode_literals import re from io import StringIO -from typing import Iterable, Optional, Union + class UnidiffParseError(Exception): ... + open_file = open make_str = str implements_to_string = lambda x: x @@ -38,65 +39,77 @@ class UnidiffParseError(Exception): ... basestring = str RE_SOURCE_FILENAME = re.compile( - r'^--- (?P"?[^\t\n]+"?)(?:\t(?P[^\n]+))?') + r'^--- (?P"?[^\t\n]+"?)(?:\t(?P[^\n]+))?' +) RE_TARGET_FILENAME = re.compile( - r'^\+\+\+ (?P"?[^\t\n]+"?)(?:\t(?P[^\n]+))?') + r'^\+\+\+ (?P"?[^\t\n]+"?)(?:\t(?P[^\n]+))?' +) # check diff git line for git renamed files support RE_DIFF_GIT_HEADER = re.compile( - r'^diff --git (?P"?a/[^\t\n]+"?) (?P"?b/[^\t\n]+"?)') + r'^diff --git (?P"?a/[^\t\n]+"?) (?P"?b/[^\t\n]+"?)' +) RE_DIFF_GIT_HEADER_URI_LIKE = re.compile( - r'^diff --git (?P.*://[^\t\n]+) (?P.*://[^\t\n]+)') + r"^diff --git (?P.*://[^\t\n]+) (?P.*://[^\t\n]+)" +) RE_DIFF_GIT_HEADER_NO_PREFIX = re.compile( - r'^diff --git (?P[^\t\n]+) (?P[^\t\n]+)') + r"^diff --git (?P[^\t\n]+) (?P[^\t\n]+)" +) # check diff git new file marker `deleted file mode 100644` -RE_DIFF_GIT_DELETED_FILE = re.compile(r'^deleted file mode \d+$') +RE_DIFF_GIT_DELETED_FILE = re.compile(r"^deleted file mode \d+$") # check diff git new file marker `new file mode 100644` -RE_DIFF_GIT_NEW_FILE = re.compile(r'^new file mode \d+$') +RE_DIFF_GIT_NEW_FILE = re.compile(r"^new file mode \d+$") # @@ (source offset, length) (target offset, length) @@ (section header) -RE_HUNK_HEADER = re.compile( - r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))?\ @@[ ]?(.*)") +RE_HUNK_HEADER = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))?\ @@[ ]?(.*)") # kept line (context) # \n empty line (treat like context) # + added line # - deleted line # \ No newline case -RE_HUNK_BODY_LINE = re.compile( - r'^(?P[- \+\\])(?P.*)', re.DOTALL) +RE_HUNK_BODY_LINE = re.compile(r"^(?P[- \+\\])(?P.*)", re.DOTALL) RE_HUNK_EMPTY_BODY_LINE = re.compile( - r'^(?P[- \+\\]?)(?P[\r\n]{1,2})', re.DOTALL) + r"^(?P[- \+\\]?)(?P[\r\n]{1,2})", re.DOTALL +) -RE_NO_NEWLINE_MARKER = re.compile(r'^\\ No newline at end of file') +RE_NO_NEWLINE_MARKER = re.compile(r"^\\ No newline at end of file") RE_BINARY_DIFF = re.compile( - r'^Binary files? ' - r'(?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?' - r'(?: and (?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?)? (differ|has changed)') + r"^Binary files? " + r"(?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?" + r"(?: and (?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?)? (differ|has changed)" +) + +DEFAULT_ENCODING = "UTF-8" -DEFAULT_ENCODING = 'UTF-8' +DEV_NULL = "/dev/null" +LINE_TYPE_ADDED = "+" +LINE_TYPE_REMOVED = "-" +LINE_TYPE_CONTEXT = " " +LINE_TYPE_EMPTY = "" +LINE_TYPE_NO_NEWLINE = "\\" +LINE_VALUE_NO_NEWLINE = " No newline at end of file" -DEV_NULL = '/dev/null' -LINE_TYPE_ADDED = '+' -LINE_TYPE_REMOVED = '-' -LINE_TYPE_CONTEXT = ' ' -LINE_TYPE_EMPTY = '' -LINE_TYPE_NO_NEWLINE = '\\' -LINE_VALUE_NO_NEWLINE = ' No newline at end of file' @implements_to_string -class Line(object): +class Line: """A diff line.""" - def __init__(self, value, line_type, - source_line_no=None, target_line_no=None, diff_line_no=None): + def __init__( + self, + value, + line_type, + source_line_no=None, + target_line_no=None, + diff_line_no=None, + ): # type: (str, str, Optional[int], Optional[int], Optional[int]) -> None - super(Line, self).__init__() + super().__init__() self.source_line_no = source_line_no self.target_line_no = target_line_no self.diff_line_no = diff_line_no @@ -113,11 +126,13 @@ def __str__(self): def __eq__(self, other): # type: (Line) -> bool - return (self.source_line_no == other.source_line_no and - self.target_line_no == other.target_line_no and - self.diff_line_no == other.diff_line_no and - self.line_type == other.line_type and - self.value == other.value) + return ( + self.source_line_no == other.source_line_no + and self.target_line_no == other.target_line_no + and self.diff_line_no == other.diff_line_no + and self.line_type == other.line_type + and self.value == other.value + ) @property def is_added(self): @@ -137,7 +152,8 @@ def is_context(self): @implements_to_string class PatchInfo(list): - """Lines with extended patch info. + """ + Lines with extended patch info. Format of this info is not documented and it very much depends on patch producer. @@ -151,17 +167,18 @@ def __repr__(self): def __str__(self): # type: () -> str - return ''.join(unicode(line) for line in self) + return "".join(unicode(line) for line in self) @implements_to_string class Hunk(list): """Each of the modified blocks of a file.""" - def __init__(self, src_start=0, src_len=0, tgt_start=0, tgt_len=0, - section_header=''): + def __init__( + self, src_start=0, src_len=0, tgt_start=0, tgt_len=0, section_header="" + ): # type: (int, int, int, int, str) -> None - super(Hunk, self).__init__() + super().__init__() if src_len is None: src_len = 1 if tgt_len is None: @@ -176,21 +193,26 @@ def __init__(self, src_start=0, src_len=0, tgt_start=0, tgt_len=0, def __repr__(self): # type: () -> str - value = "" % (self.source_start, - self.source_length, - self.target_start, - self.target_length, - self.section_header) + value = "" % ( + self.source_start, + self.source_length, + self.target_start, + self.target_length, + self.section_header, + ) return make_str(value) def __str__(self): # type: () -> str # section header is optional and thus we output it only if it's present head = "@@ -%d,%d +%d,%d @@%s\n" % ( - self.source_start, self.source_length, - self.target_start, self.target_length, - ' ' + self.section_header if self.section_header else '') - content = ''.join(unicode(line) for line in self) + self.source_start, + self.source_length, + self.target_start, + self.target_length, + " " + self.section_header if self.section_header else "", + ) + content = "".join(unicode(line) for line in self) return head + content def append(self, line): @@ -199,7 +221,7 @@ def append(self, line): # Make sure the line is encoded correctly. This is a no-op except for # potentially raising a UnicodeDecodeError. str(line) - super(Hunk, self).append(line) + super().append(line) @property def added(self): @@ -222,8 +244,10 @@ def removed(self): def is_valid(self): # type: () -> bool """Check hunk header data matches entered lines info.""" - return (len(self.source) == self.source_length and - len(self.target) == self.target_length) + return ( + len(self.source) == self.source_length + and len(self.target) == self.target_length + ) def source_lines(self): # type: () -> Iterable[Line] @@ -249,11 +273,17 @@ def target(self): class PatchedFile(list): """Patch updated file, it is a list of Hunks.""" - def __init__(self, patch_info=None, source='', target='', - source_timestamp=None, target_timestamp=None, - is_binary_file=False): + def __init__( + self, + patch_info=None, + source="", + target="", + source_timestamp=None, + target_timestamp=None, + is_binary_file=False, + ): # type: (Optional[PatchInfo], str, str, Optional[str], Optional[str], bool, bool) -> None - super(PatchedFile, self).__init__() + super().__init__() self.patch_info = patch_info self.source_file = source self.source_timestamp = source_timestamp @@ -267,18 +297,20 @@ def __repr__(self): def __str__(self): # type: () -> str - source = '' - target = '' + source = "" + target = "" # patch info is optional - info = '' if self.patch_info is None else str(self.patch_info) + info = "" if self.patch_info is None else str(self.patch_info) if not self.is_binary_file and self: source = "--- %s%s\n" % ( self.source_file, - '\t' + self.source_timestamp if self.source_timestamp else '') + "\t" + self.source_timestamp if self.source_timestamp else "", + ) target = "+++ %s%s\n" % ( self.target_file, - '\t' + self.target_timestamp if self.target_timestamp else '') - hunks = ''.join(unicode(hunk) for hunk in self) + "\t" + self.target_timestamp if self.target_timestamp else "", + ) + hunks = "".join(unicode(hunk) for hunk in self) return info + source + target + hunks def _parse_hunk(self, header, diff, encoding, metadata_only): @@ -302,12 +334,13 @@ def _parse_hunk(self, header, diff, encoding, metadata_only): if metadata_only: # quick line type detection, no regex required line_type = line[0] if line else LINE_TYPE_CONTEXT - if line_type not in (LINE_TYPE_ADDED, - LINE_TYPE_REMOVED, - LINE_TYPE_CONTEXT, - LINE_TYPE_NO_NEWLINE): - raise UnidiffParseError( - 'Hunk diff line expected: %s' % line) + if line_type not in ( + LINE_TYPE_ADDED, + LINE_TYPE_REMOVED, + LINE_TYPE_CONTEXT, + LINE_TYPE_NO_NEWLINE, + ): + raise UnidiffParseError("Hunk diff line expected: %s" % line) if line_type == LINE_TYPE_ADDED: target_line_no += 1 @@ -329,14 +362,13 @@ def _parse_hunk(self, header, diff, encoding, metadata_only): valid_line = RE_HUNK_EMPTY_BODY_LINE.match(line) if not valid_line: - raise UnidiffParseError( - 'Hunk diff line expected: %s' % line) + raise UnidiffParseError("Hunk diff line expected: %s" % line) - line_type = valid_line.group('line_type') + line_type = valid_line.group("line_type") if line_type == LINE_TYPE_EMPTY: line_type = LINE_TYPE_CONTEXT - value = valid_line.group('value') # type: str + value = valid_line.group("value") # type: str original_line = Line(value, line_type=line_type) if line_type == LINE_TYPE_ADDED: @@ -356,23 +388,26 @@ def _parse_hunk(self, header, diff, encoding, metadata_only): original_line = None # stop parsing if we got past expected number of lines - if (source_line_no > expected_source_end or - target_line_no > expected_target_end): - raise UnidiffParseError('Hunk is longer than expected') + if ( + source_line_no > expected_source_end + or target_line_no > expected_target_end + ): + raise UnidiffParseError("Hunk is longer than expected") if original_line: original_line.diff_line_no = diff_line_no hunk.append(original_line) # if hunk source/target lengths are ok, hunk is complete - if (source_line_no == expected_source_end and - target_line_no == expected_target_end): + if ( + source_line_no == expected_source_end + and target_line_no == expected_target_end + ): break # report an error if we haven't got expected number of lines - if (source_line_no < expected_source_end or - target_line_no < expected_target_end): - raise UnidiffParseError('Hunk is shorter than expected') + if source_line_no < expected_source_end or target_line_no < expected_target_end: + raise UnidiffParseError("Hunk is shorter than expected") if metadata_only: # HACK: set fixed calculated values when metadata_only is enabled @@ -384,18 +419,18 @@ def _parse_hunk(self, header, diff, encoding, metadata_only): def _add_no_newline_marker_to_last_hunk(self): # type: () -> None if not self: - raise UnidiffParseError( - 'Unexpected marker:' + LINE_VALUE_NO_NEWLINE) + raise UnidiffParseError("Unexpected marker:" + LINE_VALUE_NO_NEWLINE) last_hunk = self[-1] last_hunk.append( - Line(LINE_VALUE_NO_NEWLINE + '\n', line_type=LINE_TYPE_NO_NEWLINE)) + Line(LINE_VALUE_NO_NEWLINE + "\n", line_type=LINE_TYPE_NO_NEWLINE) + ) def _append_trailing_empty_line(self): # type: () -> None if not self: - raise UnidiffParseError('Unexpected trailing newline character') + raise UnidiffParseError("Unexpected trailing newline character") last_hunk = self[-1] - last_hunk.append(Line('\n', line_type=LINE_TYPE_EMPTY)) + last_hunk.append(Line("\n", line_type=LINE_TYPE_EMPTY)) @property def path(self): @@ -403,7 +438,8 @@ def path(self): """Return the file path abstracted from VCS.""" filepath = self.source_file if filepath in (None, DEV_NULL) or ( - self.is_rename and self.target_file not in (None, DEV_NULL)): + self.is_rename and self.target_file not in (None, DEV_NULL) + ): # if this is a rename, prefer the target filename filepath = self.target_file @@ -411,11 +447,11 @@ def path(self): if quoted: filepath = filepath[1:-1] - if filepath.startswith('a/') or filepath.startswith('b/'): + if filepath.startswith("a/") or filepath.startswith("b/"): filepath = filepath[2:] if quoted: - filepath = '"{}"'.format(filepath) + filepath = f'"{filepath}"' return filepath @@ -433,9 +469,11 @@ def removed(self): @property def is_rename(self): - return (self.source_file != DEV_NULL + return ( + self.source_file != DEV_NULL and self.target_file != DEV_NULL - and self.source_file[2:] != self.target_file[2:]) + and self.source_file[2:] != self.target_file[2:] + ) @property def is_added_file(self): @@ -443,8 +481,9 @@ def is_added_file(self): """Return True if this patch adds the file.""" if self.source_file == DEV_NULL: return True - return (len(self) == 1 and self[0].source_start == 0 and - self[0].source_length == 0) + return ( + len(self) == 1 and self[0].source_start == 0 and self[0].source_length == 0 + ) @property def is_removed_file(self): @@ -452,8 +491,9 @@ def is_removed_file(self): """Return True if this patch removes the file.""" if self.target_file == DEV_NULL: return True - return (len(self) == 1 and self[0].target_start == 0 and - self[0].target_length == 0) + return ( + len(self) == 1 and self[0].target_start == 0 and self[0].target_length == 0 + ) @property def is_modified_file(self): @@ -468,7 +508,7 @@ class PatchSet(list): def __init__(self, f, encoding=None, metadata_only=False): # type: (Union[StringIO, str], Optional[str], bool) -> None - super(PatchSet, self).__init__() + super().__init__() # convert string inputs to StringIO objects if isinstance(f, basestring): @@ -484,11 +524,11 @@ def __init__(self, f, encoding=None, metadata_only=False): def __repr__(self): # type: () -> str - return make_str('') % super(PatchSet, self).__repr__() + return make_str("") % super().__repr__() def __str__(self): # type: () -> str - return ''.join(unicode(patched_file) for patched_file in self) + return "".join(unicode(patched_file) for patched_file in self) def _parse(self, diff, encoding, metadata_only): # type: (StringIO, Optional[str], bool) -> None @@ -501,15 +541,18 @@ def _parse(self, diff, encoding, metadata_only): line = line.decode(encoding) # check for a git file rename - is_diff_git_header = RE_DIFF_GIT_HEADER.match(line) or \ - RE_DIFF_GIT_HEADER_URI_LIKE.match(line) or \ - RE_DIFF_GIT_HEADER_NO_PREFIX.match(line) + is_diff_git_header = ( + RE_DIFF_GIT_HEADER.match(line) + or RE_DIFF_GIT_HEADER_URI_LIKE.match(line) + or RE_DIFF_GIT_HEADER_NO_PREFIX.match(line) + ) if is_diff_git_header: patch_info = PatchInfo() - source_file = is_diff_git_header.group('source') - target_file = is_diff_git_header.group('target') + source_file = is_diff_git_header.group("source") + target_file = is_diff_git_header.group("target") current_file = PatchedFile( - patch_info, source_file, target_file, None, None) + patch_info, source_file, target_file, None, None + ) self.append(current_file) patch_info.append(line) continue @@ -518,7 +561,7 @@ def _parse(self, diff, encoding, metadata_only): is_diff_git_new_file = RE_DIFF_GIT_NEW_FILE.match(line) if is_diff_git_new_file: if current_file is None or patch_info is None: - raise UnidiffParseError('Unexpected new file found: %s' % line) + raise UnidiffParseError("Unexpected new file found: %s" % line) current_file.source_file = DEV_NULL patch_info.append(line) continue @@ -527,7 +570,7 @@ def _parse(self, diff, encoding, metadata_only): is_diff_git_deleted_file = RE_DIFF_GIT_DELETED_FILE.match(line) if is_diff_git_deleted_file: if current_file is None or patch_info is None: - raise UnidiffParseError('Unexpected deleted file found: %s' % line) + raise UnidiffParseError("Unexpected deleted file found: %s" % line) current_file.target_file = DEV_NULL patch_info.append(line) continue @@ -535,12 +578,13 @@ def _parse(self, diff, encoding, metadata_only): # check for source file header is_source_filename = RE_SOURCE_FILENAME.match(line) if is_source_filename: - source_file = is_source_filename.group('filename') - source_timestamp = is_source_filename.group('timestamp') + source_file = is_source_filename.group("filename") + source_timestamp = is_source_filename.group("timestamp") # reset current file, unless we are processing a rename # (in that case, source files should match) if current_file is not None and not ( - current_file.source_file == source_file): + current_file.source_file == source_file + ): current_file = None elif current_file is not None: current_file.source_timestamp = source_timestamp @@ -549,15 +593,21 @@ def _parse(self, diff, encoding, metadata_only): # check for target file header is_target_filename = RE_TARGET_FILENAME.match(line) if is_target_filename: - target_file = is_target_filename.group('filename') - target_timestamp = is_target_filename.group('timestamp') - if current_file is not None and not (current_file.target_file == target_file): - raise UnidiffParseError('Target without source: %s' % line) + target_file = is_target_filename.group("filename") + target_timestamp = is_target_filename.group("timestamp") + if current_file is not None and not ( + current_file.target_file == target_file + ): + raise UnidiffParseError("Target without source: %s" % line) if current_file is None: # add current file to PatchSet current_file = PatchedFile( - patch_info, source_file, target_file, - source_timestamp, target_timestamp) + patch_info, + source_file, + target_file, + source_timestamp, + target_timestamp, + ) self.append(current_file) patch_info = None else: @@ -569,7 +619,7 @@ def _parse(self, diff, encoding, metadata_only): if is_hunk_header: patch_info = None if current_file is None: - raise UnidiffParseError('Unexpected hunk found: %s' % line) + raise UnidiffParseError("Unexpected hunk found: %s" % line) current_file._parse_hunk(line, diff, encoding, metadata_only) continue @@ -577,12 +627,12 @@ def _parse(self, diff, encoding, metadata_only): is_no_newline = RE_NO_NEWLINE_MARKER.match(line) if is_no_newline: if current_file is None: - raise UnidiffParseError('Unexpected marker: %s' % line) + raise UnidiffParseError("Unexpected marker: %s" % line) current_file._add_no_newline_marker_to_last_hunk() continue # sometimes hunks can be followed by empty lines - if line == '\n' and current_file is not None: + if line == "\n" and current_file is not None: current_file._append_trailing_empty_line() continue @@ -593,20 +643,21 @@ def _parse(self, diff, encoding, metadata_only): is_binary_diff = RE_BINARY_DIFF.match(line) if is_binary_diff: - source_file = is_binary_diff.group('source_filename') - target_file = is_binary_diff.group('target_filename') + source_file = is_binary_diff.group("source_filename") + target_file = is_binary_diff.group("target_filename") patch_info.append(line) if current_file is not None: current_file.is_binary_file = True else: current_file = PatchedFile( - patch_info, source_file, target_file, is_binary_file=True) + patch_info, source_file, target_file, is_binary_file=True + ) self.append(current_file) patch_info = None current_file = None continue - if line == 'GIT binary patch\n': + if line == "GIT binary patch\n": current_file.is_binary_file = True patch_info = None current_file = None @@ -615,15 +666,19 @@ def _parse(self, diff, encoding, metadata_only): patch_info.append(line) @classmethod - def from_filename(cls, filename, encoding=DEFAULT_ENCODING, errors=None, newline=None): + def from_filename( + cls, filename, encoding=DEFAULT_ENCODING, errors=None, newline=None + ): # type: (str, str, Optional[str]) -> PatchSet """Return a PatchSet instance given a diff filename.""" - with open_file(filename, 'r', encoding=encoding, errors=errors, newline=newline) as f: + with open_file( + filename, "r", encoding=encoding, errors=errors, newline=newline + ) as f: instance = cls(f) return instance @staticmethod - def _convert_string(data, encoding=None, errors='strict'): + def _convert_string(data, encoding=None, errors="strict"): # type: (Union[str, bytes], str, str) -> StringIO if encoding is not None: # if encoding is given, assume bytes and decode @@ -631,7 +686,7 @@ def _convert_string(data, encoding=None, errors='strict'): return StringIO(data) @classmethod - def from_string(cls, data, encoding=None, errors='strict'): + def from_string(cls, data, encoding=None, errors="strict"): # type: (str, str, Optional[str]) -> PatchSet """Return a PatchSet instance given a diff string.""" return cls(cls._convert_string(data, encoding, errors)) @@ -664,4 +719,4 @@ def added(self): def removed(self): # type: () -> int """Return the patch total removed lines.""" - return sum([f.removed for f in self]) \ No newline at end of file + return sum([f.removed for f in self]) diff --git a/scanpipe/pipes/vulnerablecode.py b/scanpipe/pipes/vulnerablecode.py index 4deaefc704..57924bff70 100644 --- a/scanpipe/pipes/vulnerablecode.py +++ b/scanpipe/pipes/vulnerablecode.py @@ -109,6 +109,7 @@ def request_post( def bulk_search_by_purl( purls, + reachability=False, timeout=None, api_url=VULNERABLECODE_API_URL, ): @@ -118,7 +119,7 @@ def bulk_search_by_purl( data = { "purls": purls, "details": True, - "reachability": True, + "reachability": reachability, } logger.debug(f"VulnerableCode: url={url} purls_count={len(purls)}") @@ -137,7 +138,7 @@ def filter_vulnerabilities(vulnerabilities, ignore_set): def fetch_vulnerabilities( - packages, chunk_size=1000, logger=logger.info, ignore_set=None + packages, chunk_size=1000, logger=logger.info, ignore_set=None, reachability=False ): """ Fetch and store vulnerabilities for each provided `packages`. @@ -147,7 +148,7 @@ def fetch_vulnerabilities( for purls_batch in chunked(get_purls(packages), chunk_size): try: - response_data = bulk_search_by_purl(purls_batch) + response_data = bulk_search_by_purl(purls_batch, reachability=reachability) except (requests.RequestException, ValueError, TypeError) as exception: logger(f"{label} [Exception] {exception}") return From c58c1f121c07f05ebc60b8f9a2957f4c2322ba81 Mon Sep 17 00:00:00 2001 From: ziad hany Date: Thu, 23 Jul 2026 03:53:25 +0300 Subject: [PATCH 05/13] Fix CI files format Signed-off-by: ziad hany --- scanpipe/pipes/unidiff/patch.py | 75 ++++++++++++++++----------------- 1 file changed, 37 insertions(+), 38 deletions(-) diff --git a/scanpipe/pipes/unidiff/patch.py b/scanpipe/pipes/unidiff/patch.py index 8075a8cc77..e53c7c7d94 100644 --- a/scanpipe/pipes/unidiff/patch.py +++ b/scanpipe/pipes/unidiff/patch.py @@ -34,7 +34,12 @@ class UnidiffParseError(Exception): ... open_file = open make_str = str -implements_to_string = lambda x: x + + +def implements_to_string(x): + return x + + unicode = str basestring = str @@ -82,7 +87,8 @@ class UnidiffParseError(Exception): ... RE_BINARY_DIFF = re.compile( r"^Binary files? " r"(?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?" - r"(?: and (?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?)? (differ|has changed)" + r"(?: and (?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?)?" + r" (differ|has changed)" ) DEFAULT_ENCODING = "UTF-8" @@ -122,7 +128,7 @@ def __repr__(self): def __str__(self): # type: () -> str - return "%s%s" % (self.line_type, self.value) + return f"{self.line_type}{self.value}" def __eq__(self, other): # type: (Line) -> bool @@ -162,7 +168,7 @@ class PatchInfo(list): def __repr__(self): # type: () -> str - value = "" % self[0].strip() + value = f"" return make_str(value) def __str__(self): @@ -193,24 +199,19 @@ def __init__( def __repr__(self): # type: () -> str - value = "" % ( - self.source_start, - self.source_length, - self.target_start, - self.target_length, - self.section_header, + value = ( + f"" ) return make_str(value) def __str__(self): # type: () -> str # section header is optional and thus we output it only if it's present - head = "@@ -%d,%d +%d,%d @@%s\n" % ( - self.source_start, - self.source_length, - self.target_start, - self.target_length, - " " + self.section_header if self.section_header else "", + section_hdr = f" {self.section_header}" if self.section_header else "" + head = ( + f"@@ -{self.source_start},{self.source_length} " + f"+{self.target_start},{self.target_length} @@{section_hdr}\n" ) content = "".join(unicode(line) for line in self) return head + content @@ -252,22 +253,22 @@ def is_valid(self): def source_lines(self): # type: () -> Iterable[Line] """Hunk lines from source file (generator).""" - return (l for l in self if l.is_context or l.is_removed) + return (line for line in self if line.is_context or line.is_removed) @property def source(self): # type: () -> Iterable[str] - return [str(l) for l in self.source_lines()] + return [str(line) for line in self.source_lines()] def target_lines(self): # type: () -> Iterable[Line] """Hunk lines from target file (generator).""" - return (l for l in self if l.is_context or l.is_added) + return (line for line in self if line.is_context or line.is_added) @property def target(self): # type: () -> Iterable[str] - return [str(l) for l in self.target_lines()] + return [str(line) for line in self.target_lines()] class PatchedFile(list): @@ -302,18 +303,16 @@ def __str__(self): # patch info is optional info = "" if self.patch_info is None else str(self.patch_info) if not self.is_binary_file and self: - source = "--- %s%s\n" % ( - self.source_file, - "\t" + self.source_timestamp if self.source_timestamp else "", - ) - target = "+++ %s%s\n" % ( - self.target_file, - "\t" + self.target_timestamp if self.target_timestamp else "", - ) + source_ts = f"\t{self.source_timestamp}" if self.source_timestamp else "" + source = f"--- {self.source_file}{source_ts}\n" + + target_ts = f"\t{self.target_timestamp}" if self.target_timestamp else "" + target = f"+++ {self.target_file}{target_ts}\n" + hunks = "".join(unicode(hunk) for hunk in self) return info + source + target + hunks - def _parse_hunk(self, header, diff, encoding, metadata_only): + def _parse_hunk(self, header, diff, encoding, metadata_only): # noqa: C901 # type: (str, enumerate[str], Optional[str], bool) -> None """Parse hunk details.""" header_info = RE_HUNK_HEADER.match(header) @@ -340,7 +339,7 @@ def _parse_hunk(self, header, diff, encoding, metadata_only): LINE_TYPE_CONTEXT, LINE_TYPE_NO_NEWLINE, ): - raise UnidiffParseError("Hunk diff line expected: %s" % line) + raise UnidiffParseError(f"Hunk diff line expected: {line}") if line_type == LINE_TYPE_ADDED: target_line_no += 1 @@ -362,7 +361,7 @@ def _parse_hunk(self, header, diff, encoding, metadata_only): valid_line = RE_HUNK_EMPTY_BODY_LINE.match(line) if not valid_line: - raise UnidiffParseError("Hunk diff line expected: %s" % line) + raise UnidiffParseError(f"Hunk diff line expected: {line}") line_type = valid_line.group("line_type") if line_type == LINE_TYPE_EMPTY: @@ -410,7 +409,7 @@ def _parse_hunk(self, header, diff, encoding, metadata_only): raise UnidiffParseError("Hunk is shorter than expected") if metadata_only: - # HACK: set fixed calculated values when metadata_only is enabled + # set fixed calculated values when metadata_only is enabled hunk._added = added hunk._removed = removed @@ -530,7 +529,7 @@ def __str__(self): # type: () -> str return "".join(unicode(patched_file) for patched_file in self) - def _parse(self, diff, encoding, metadata_only): + def _parse(self, diff, encoding, metadata_only): # noqa: C901 # type: (StringIO, Optional[str], bool) -> None current_file = None patch_info = None @@ -561,7 +560,7 @@ def _parse(self, diff, encoding, metadata_only): is_diff_git_new_file = RE_DIFF_GIT_NEW_FILE.match(line) if is_diff_git_new_file: if current_file is None or patch_info is None: - raise UnidiffParseError("Unexpected new file found: %s" % line) + raise UnidiffParseError(f"Unexpected new file found: {line}") current_file.source_file = DEV_NULL patch_info.append(line) continue @@ -570,7 +569,7 @@ def _parse(self, diff, encoding, metadata_only): is_diff_git_deleted_file = RE_DIFF_GIT_DELETED_FILE.match(line) if is_diff_git_deleted_file: if current_file is None or patch_info is None: - raise UnidiffParseError("Unexpected deleted file found: %s" % line) + raise UnidiffParseError(f"Unexpected deleted file found: {line}") current_file.target_file = DEV_NULL patch_info.append(line) continue @@ -598,7 +597,7 @@ def _parse(self, diff, encoding, metadata_only): if current_file is not None and not ( current_file.target_file == target_file ): - raise UnidiffParseError("Target without source: %s" % line) + raise UnidiffParseError(f"Target without source: {line}") if current_file is None: # add current file to PatchSet current_file = PatchedFile( @@ -619,7 +618,7 @@ def _parse(self, diff, encoding, metadata_only): if is_hunk_header: patch_info = None if current_file is None: - raise UnidiffParseError("Unexpected hunk found: %s" % line) + raise UnidiffParseError(f"Unexpected hunk found: {line}") current_file._parse_hunk(line, diff, encoding, metadata_only) continue @@ -627,7 +626,7 @@ def _parse(self, diff, encoding, metadata_only): is_no_newline = RE_NO_NEWLINE_MARKER.match(line) if is_no_newline: if current_file is None: - raise UnidiffParseError("Unexpected marker: %s" % line) + raise UnidiffParseError(f"Unexpected marker: {line}") current_file._add_no_newline_marker_to_last_hunk() continue From b344bdcc58459c8673611724a3a1c648c95baea4 Mon Sep 17 00:00:00 2001 From: ziad hany Date: Thu, 23 Jul 2026 04:02:18 +0300 Subject: [PATCH 06/13] Fix a typo in patch.py.ABOUT file Signed-off-by: ziad hany --- scanpipe/pipes/unidiff/patch.py.ABOUT | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scanpipe/pipes/unidiff/patch.py.ABOUT b/scanpipe/pipes/unidiff/patch.py.ABOUT index 3118ed9643..75c3bb8ade 100644 --- a/scanpipe/pipes/unidiff/patch.py.ABOUT +++ b/scanpipe/pipes/unidiff/patch.py.ABOUT @@ -10,4 +10,4 @@ package_url: pkg:pypi/unidiff@0.7.5 licenses: - key: mit name: MIT License - file: mit.LICENSE \ No newline at end of file + file: patch.py.LICENSE \ No newline at end of file From a8bd834b24dcc5ffd9aaecd9f1fec77c1c108fca Mon Sep 17 00:00:00 2001 From: ziad hany Date: Wed, 29 Jul 2026 03:17:06 +0300 Subject: [PATCH 07/13] Allow reachability by default, for vulnerabilities pipeline Signed-off-by: ziad hany --- scanpipe/pipelines/find_vulnerabilities.py | 12 +----------- scanpipe/pipes/vulnerablecode.py | 7 +++---- 2 files changed, 4 insertions(+), 15 deletions(-) diff --git a/scanpipe/pipelines/find_vulnerabilities.py b/scanpipe/pipelines/find_vulnerabilities.py index 1bdf8af6bc..b0c8066b9a 100644 --- a/scanpipe/pipelines/find_vulnerabilities.py +++ b/scanpipe/pipelines/find_vulnerabilities.py @@ -19,7 +19,7 @@ # # 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 aboutcode.pipeline import optional_step + from scanpipe.pipelines import Pipeline from scanpipe.pipes import vulnerablecode @@ -34,13 +34,11 @@ class FindVulnerabilities(Pipeline): download_inputs = False is_addon = True results_url = "/project/{slug}/packages/?is_vulnerable=yes" - reachability = False @classmethod def steps(cls): return ( cls.check_vulnerablecode_service_availability, - cls.enable_reachability_analysis, cls.lookup_packages_vulnerabilities, cls.lookup_dependencies_vulnerabilities, ) @@ -50,12 +48,6 @@ def get_availability(cls): if not vulnerablecode.is_configured(): return "VulnerableCode is not configured." - @optional_step("reachability") - def enable_reachability_analysis(self): - """Enable the reachability flag for vulnerability lookups.""" - self.reachability = True - self.log("Reachability analysis is ENABLED.") - def check_vulnerablecode_service_availability(self): """Check if the VulnerableCode service if configured and available.""" if not vulnerablecode.is_configured(): @@ -70,7 +62,6 @@ def lookup_packages_vulnerabilities(self): vulnerablecode.fetch_vulnerabilities( packages=packages, ignore_set=self.project.ignored_vulnerabilities_set, - reachability=self.reachability, logger=self.log, ) @@ -80,6 +71,5 @@ def lookup_dependencies_vulnerabilities(self): vulnerablecode.fetch_vulnerabilities( packages=dependencies, ignore_set=self.project.ignored_vulnerabilities_set, - reachability=self.reachability, logger=self.log, ) diff --git a/scanpipe/pipes/vulnerablecode.py b/scanpipe/pipes/vulnerablecode.py index 57924bff70..4deaefc704 100644 --- a/scanpipe/pipes/vulnerablecode.py +++ b/scanpipe/pipes/vulnerablecode.py @@ -109,7 +109,6 @@ def request_post( def bulk_search_by_purl( purls, - reachability=False, timeout=None, api_url=VULNERABLECODE_API_URL, ): @@ -119,7 +118,7 @@ def bulk_search_by_purl( data = { "purls": purls, "details": True, - "reachability": reachability, + "reachability": True, } logger.debug(f"VulnerableCode: url={url} purls_count={len(purls)}") @@ -138,7 +137,7 @@ def filter_vulnerabilities(vulnerabilities, ignore_set): def fetch_vulnerabilities( - packages, chunk_size=1000, logger=logger.info, ignore_set=None, reachability=False + packages, chunk_size=1000, logger=logger.info, ignore_set=None ): """ Fetch and store vulnerabilities for each provided `packages`. @@ -148,7 +147,7 @@ def fetch_vulnerabilities( for purls_batch in chunked(get_purls(packages), chunk_size): try: - response_data = bulk_search_by_purl(purls_batch, reachability=reachability) + response_data = bulk_search_by_purl(purls_batch) except (requests.RequestException, ValueError, TypeError) as exception: logger(f"{label} [Exception] {exception}") return From 7cd37b28bdb8f49289ec4d38b4f291891a2f175e Mon Sep 17 00:00:00 2001 From: ziad hany Date: Thu, 30 Jul 2026 02:23:55 +0300 Subject: [PATCH 08/13] Update the code to difflib instead of unidiff library Fix a bug related to constant detections and add a test Signed-off-by: ziad hany --- scanpipe/pipes/unidiff/patch.py | 721 ------------------ scanpipe/pipes/unidiff/patch.py.ABOUT | 13 - scanpipe/pipes/unidiff/patch.py.LICENSE | 20 - scanpipe/tests/data/reachability/app.py | 2 + .../tests/data/reachability/diff-app.patch | 39 - scanpipe/tests/data/reachability/fixed-app.py | 2 + scanpipe/tests/data/reachability/vuln-app.py | 2 + 7 files changed, 6 insertions(+), 793 deletions(-) delete mode 100644 scanpipe/pipes/unidiff/patch.py delete mode 100644 scanpipe/pipes/unidiff/patch.py.ABOUT delete mode 100644 scanpipe/pipes/unidiff/patch.py.LICENSE delete mode 100644 scanpipe/tests/data/reachability/diff-app.patch diff --git a/scanpipe/pipes/unidiff/patch.py b/scanpipe/pipes/unidiff/patch.py deleted file mode 100644 index e53c7c7d94..0000000000 --- a/scanpipe/pipes/unidiff/patch.py +++ /dev/null @@ -1,721 +0,0 @@ -# Extracted essential patch code analyzer and modified -# from the original unidiff library: -# https://github.com/matiasb/python-unidiff/blob/2771a878f7bc6619e625feb4dbad3427f57f5237/unidiff/patch.py - -# -*- coding: utf-8 -*- - -# The MIT License (MIT) -# Copyright (c) 2014-2023 Matias Bordese -# -# Permission is hereby granted, free of charge, to any person obtaining a copy -# of this software and associated documentation files (the "Software"), to deal -# in the Software without restriction, including without limitation the rights -# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -# copies of the Software, and to permit persons to whom the Software is -# furnished to do so, subject to the following conditions: -# -# The above copyright notice and this permission notice shall be included in -# all copies or substantial portions of the Software. -# -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, -# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF -# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. -# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, -# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR -# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE -# OR OTHER DEALINGS IN THE SOFTWARE. - -import re -from io import StringIO - - -class UnidiffParseError(Exception): ... - - -open_file = open -make_str = str - - -def implements_to_string(x): - return x - - -unicode = str -basestring = str - -RE_SOURCE_FILENAME = re.compile( - r'^--- (?P"?[^\t\n]+"?)(?:\t(?P[^\n]+))?' -) -RE_TARGET_FILENAME = re.compile( - r'^\+\+\+ (?P"?[^\t\n]+"?)(?:\t(?P[^\n]+))?' -) - - -# check diff git line for git renamed files support -RE_DIFF_GIT_HEADER = re.compile( - r'^diff --git (?P"?a/[^\t\n]+"?) (?P"?b/[^\t\n]+"?)' -) -RE_DIFF_GIT_HEADER_URI_LIKE = re.compile( - r"^diff --git (?P.*://[^\t\n]+) (?P.*://[^\t\n]+)" -) -RE_DIFF_GIT_HEADER_NO_PREFIX = re.compile( - r"^diff --git (?P[^\t\n]+) (?P[^\t\n]+)" -) - -# check diff git new file marker `deleted file mode 100644` -RE_DIFF_GIT_DELETED_FILE = re.compile(r"^deleted file mode \d+$") - -# check diff git new file marker `new file mode 100644` -RE_DIFF_GIT_NEW_FILE = re.compile(r"^new file mode \d+$") - - -# @@ (source offset, length) (target offset, length) @@ (section header) -RE_HUNK_HEADER = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))?\ @@[ ]?(.*)") - -# kept line (context) -# \n empty line (treat like context) -# + added line -# - deleted line -# \ No newline case -RE_HUNK_BODY_LINE = re.compile(r"^(?P[- \+\\])(?P.*)", re.DOTALL) -RE_HUNK_EMPTY_BODY_LINE = re.compile( - r"^(?P[- \+\\]?)(?P[\r\n]{1,2})", re.DOTALL -) - -RE_NO_NEWLINE_MARKER = re.compile(r"^\\ No newline at end of file") - -RE_BINARY_DIFF = re.compile( - r"^Binary files? " - r"(?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?" - r"(?: and (?P[^\t]+?)(?:\t(?P[\s0-9:\+-]+))?)?" - r" (differ|has changed)" -) - -DEFAULT_ENCODING = "UTF-8" - -DEV_NULL = "/dev/null" -LINE_TYPE_ADDED = "+" -LINE_TYPE_REMOVED = "-" -LINE_TYPE_CONTEXT = " " -LINE_TYPE_EMPTY = "" -LINE_TYPE_NO_NEWLINE = "\\" -LINE_VALUE_NO_NEWLINE = " No newline at end of file" - - -@implements_to_string -class Line: - """A diff line.""" - - def __init__( - self, - value, - line_type, - source_line_no=None, - target_line_no=None, - diff_line_no=None, - ): - # type: (str, str, Optional[int], Optional[int], Optional[int]) -> None - super().__init__() - self.source_line_no = source_line_no - self.target_line_no = target_line_no - self.diff_line_no = diff_line_no - self.line_type = line_type - self.value = value - - def __repr__(self): - # type: () -> str - return make_str("") % (self.line_type, self.value) - - def __str__(self): - # type: () -> str - return f"{self.line_type}{self.value}" - - def __eq__(self, other): - # type: (Line) -> bool - return ( - self.source_line_no == other.source_line_no - and self.target_line_no == other.target_line_no - and self.diff_line_no == other.diff_line_no - and self.line_type == other.line_type - and self.value == other.value - ) - - @property - def is_added(self): - # type: () -> bool - return self.line_type == LINE_TYPE_ADDED - - @property - def is_removed(self): - # type: () -> bool - return self.line_type == LINE_TYPE_REMOVED - - @property - def is_context(self): - # type: () -> bool - return self.line_type == LINE_TYPE_CONTEXT - - -@implements_to_string -class PatchInfo(list): - """ - Lines with extended patch info. - - Format of this info is not documented and it very much depends on - patch producer. - - """ - - def __repr__(self): - # type: () -> str - value = f"" - return make_str(value) - - def __str__(self): - # type: () -> str - return "".join(unicode(line) for line in self) - - -@implements_to_string -class Hunk(list): - """Each of the modified blocks of a file.""" - - def __init__( - self, src_start=0, src_len=0, tgt_start=0, tgt_len=0, section_header="" - ): - # type: (int, int, int, int, str) -> None - super().__init__() - if src_len is None: - src_len = 1 - if tgt_len is None: - tgt_len = 1 - self.source_start = int(src_start) - self.source_length = int(src_len) - self.target_start = int(tgt_start) - self.target_length = int(tgt_len) - self.section_header = section_header - self._added = None # Optional[int] - self._removed = None # Optional[int] - - def __repr__(self): - # type: () -> str - value = ( - f"" - ) - return make_str(value) - - def __str__(self): - # type: () -> str - # section header is optional and thus we output it only if it's present - section_hdr = f" {self.section_header}" if self.section_header else "" - head = ( - f"@@ -{self.source_start},{self.source_length} " - f"+{self.target_start},{self.target_length} @@{section_hdr}\n" - ) - content = "".join(unicode(line) for line in self) - return head + content - - def append(self, line): - # type: (Line) -> None - """Append the line to hunk, and keep track of source/target lines.""" - # Make sure the line is encoded correctly. This is a no-op except for - # potentially raising a UnicodeDecodeError. - str(line) - super().append(line) - - @property - def added(self): - # type: () -> Optional[int] - if self._added is not None: - return self._added - # re-calculate each time to allow for hunk modifications - # (which should mean metadata_only switch wasn't used) - return sum(1 for line in self if line.is_added) - - @property - def removed(self): - # type: () -> Optional[int] - if self._removed is not None: - return self._removed - # re-calculate each time to allow for hunk modifications - # (which should mean metadata_only switch wasn't used) - return sum(1 for line in self if line.is_removed) - - def is_valid(self): - # type: () -> bool - """Check hunk header data matches entered lines info.""" - return ( - len(self.source) == self.source_length - and len(self.target) == self.target_length - ) - - def source_lines(self): - # type: () -> Iterable[Line] - """Hunk lines from source file (generator).""" - return (line for line in self if line.is_context or line.is_removed) - - @property - def source(self): - # type: () -> Iterable[str] - return [str(line) for line in self.source_lines()] - - def target_lines(self): - # type: () -> Iterable[Line] - """Hunk lines from target file (generator).""" - return (line for line in self if line.is_context or line.is_added) - - @property - def target(self): - # type: () -> Iterable[str] - return [str(line) for line in self.target_lines()] - - -class PatchedFile(list): - """Patch updated file, it is a list of Hunks.""" - - def __init__( - self, - patch_info=None, - source="", - target="", - source_timestamp=None, - target_timestamp=None, - is_binary_file=False, - ): - # type: (Optional[PatchInfo], str, str, Optional[str], Optional[str], bool, bool) -> None - super().__init__() - self.patch_info = patch_info - self.source_file = source - self.source_timestamp = source_timestamp - self.target_file = target - self.target_timestamp = target_timestamp - self.is_binary_file = is_binary_file - - def __repr__(self): - # type: () -> str - return make_str("") % make_str(self.path) - - def __str__(self): - # type: () -> str - source = "" - target = "" - # patch info is optional - info = "" if self.patch_info is None else str(self.patch_info) - if not self.is_binary_file and self: - source_ts = f"\t{self.source_timestamp}" if self.source_timestamp else "" - source = f"--- {self.source_file}{source_ts}\n" - - target_ts = f"\t{self.target_timestamp}" if self.target_timestamp else "" - target = f"+++ {self.target_file}{target_ts}\n" - - hunks = "".join(unicode(hunk) for hunk in self) - return info + source + target + hunks - - def _parse_hunk(self, header, diff, encoding, metadata_only): # noqa: C901 - # type: (str, enumerate[str], Optional[str], bool) -> None - """Parse hunk details.""" - header_info = RE_HUNK_HEADER.match(header) - hunk_info = header_info.groups() - hunk = Hunk(*hunk_info) - - source_line_no = hunk.source_start - target_line_no = hunk.target_start - expected_source_end = source_line_no + hunk.source_length - expected_target_end = target_line_no + hunk.target_length - added = 0 - removed = 0 - - for diff_line_no, line in diff: - if encoding is not None: - line = line.decode(encoding) - - if metadata_only: - # quick line type detection, no regex required - line_type = line[0] if line else LINE_TYPE_CONTEXT - if line_type not in ( - LINE_TYPE_ADDED, - LINE_TYPE_REMOVED, - LINE_TYPE_CONTEXT, - LINE_TYPE_NO_NEWLINE, - ): - raise UnidiffParseError(f"Hunk diff line expected: {line}") - - if line_type == LINE_TYPE_ADDED: - target_line_no += 1 - added += 1 - elif line_type == LINE_TYPE_REMOVED: - source_line_no += 1 - removed += 1 - elif line_type == LINE_TYPE_CONTEXT: - target_line_no += 1 - source_line_no += 1 - - # no file content tracking - original_line = None - - else: - # parse diff line content - valid_line = RE_HUNK_BODY_LINE.match(line) - if not valid_line: - valid_line = RE_HUNK_EMPTY_BODY_LINE.match(line) - - if not valid_line: - raise UnidiffParseError(f"Hunk diff line expected: {line}") - - line_type = valid_line.group("line_type") - if line_type == LINE_TYPE_EMPTY: - line_type = LINE_TYPE_CONTEXT - - value = valid_line.group("value") # type: str - original_line = Line(value, line_type=line_type) - - if line_type == LINE_TYPE_ADDED: - original_line.target_line_no = target_line_no - target_line_no += 1 - elif line_type == LINE_TYPE_REMOVED: - original_line.source_line_no = source_line_no - source_line_no += 1 - elif line_type == LINE_TYPE_CONTEXT: - original_line.target_line_no = target_line_no - original_line.source_line_no = source_line_no - target_line_no += 1 - source_line_no += 1 - elif line_type == LINE_TYPE_NO_NEWLINE: - pass - else: - original_line = None - - # stop parsing if we got past expected number of lines - if ( - source_line_no > expected_source_end - or target_line_no > expected_target_end - ): - raise UnidiffParseError("Hunk is longer than expected") - - if original_line: - original_line.diff_line_no = diff_line_no - hunk.append(original_line) - - # if hunk source/target lengths are ok, hunk is complete - if ( - source_line_no == expected_source_end - and target_line_no == expected_target_end - ): - break - - # report an error if we haven't got expected number of lines - if source_line_no < expected_source_end or target_line_no < expected_target_end: - raise UnidiffParseError("Hunk is shorter than expected") - - if metadata_only: - # set fixed calculated values when metadata_only is enabled - hunk._added = added - hunk._removed = removed - - self.append(hunk) - - def _add_no_newline_marker_to_last_hunk(self): - # type: () -> None - if not self: - raise UnidiffParseError("Unexpected marker:" + LINE_VALUE_NO_NEWLINE) - last_hunk = self[-1] - last_hunk.append( - Line(LINE_VALUE_NO_NEWLINE + "\n", line_type=LINE_TYPE_NO_NEWLINE) - ) - - def _append_trailing_empty_line(self): - # type: () -> None - if not self: - raise UnidiffParseError("Unexpected trailing newline character") - last_hunk = self[-1] - last_hunk.append(Line("\n", line_type=LINE_TYPE_EMPTY)) - - @property - def path(self): - # type: () -> str - """Return the file path abstracted from VCS.""" - filepath = self.source_file - if filepath in (None, DEV_NULL) or ( - self.is_rename and self.target_file not in (None, DEV_NULL) - ): - # if this is a rename, prefer the target filename - filepath = self.target_file - - quoted = filepath.startswith('"') and filepath.endswith('"') - if quoted: - filepath = filepath[1:-1] - - if filepath.startswith("a/") or filepath.startswith("b/"): - filepath = filepath[2:] - - if quoted: - filepath = f'"{filepath}"' - - return filepath - - @property - def added(self): - # type: () -> int - """Return the file total added lines.""" - return sum([hunk.added for hunk in self]) - - @property - def removed(self): - # type: () -> int - """Return the file total removed lines.""" - return sum([hunk.removed for hunk in self]) - - @property - def is_rename(self): - return ( - self.source_file != DEV_NULL - and self.target_file != DEV_NULL - and self.source_file[2:] != self.target_file[2:] - ) - - @property - def is_added_file(self): - # type: () -> bool - """Return True if this patch adds the file.""" - if self.source_file == DEV_NULL: - return True - return ( - len(self) == 1 and self[0].source_start == 0 and self[0].source_length == 0 - ) - - @property - def is_removed_file(self): - # type: () -> bool - """Return True if this patch removes the file.""" - if self.target_file == DEV_NULL: - return True - return ( - len(self) == 1 and self[0].target_start == 0 and self[0].target_length == 0 - ) - - @property - def is_modified_file(self): - # type: () -> bool - """Return True if this patch modifies the file.""" - return not (self.is_added_file or self.is_removed_file) - - -@implements_to_string -class PatchSet(list): - """A list of PatchedFiles.""" - - def __init__(self, f, encoding=None, metadata_only=False): - # type: (Union[StringIO, str], Optional[str], bool) -> None - super().__init__() - - # convert string inputs to StringIO objects - if isinstance(f, basestring): - f = self._convert_string(f, encoding) # type: StringIO - - # make sure we pass an iterator object to parse - data = iter(f) - # if encoding is None, assume we are reading unicode data - # when metadata_only is True, only perform a minimal metadata parsing - # (ie. hunks without content) which is around 2.5-6 times faster; - # it will still validate the diff metadata consistency and get counts - self._parse(data, encoding=encoding, metadata_only=metadata_only) - - def __repr__(self): - # type: () -> str - return make_str("") % super().__repr__() - - def __str__(self): - # type: () -> str - return "".join(unicode(patched_file) for patched_file in self) - - def _parse(self, diff, encoding, metadata_only): # noqa: C901 - # type: (StringIO, Optional[str], bool) -> None - current_file = None - patch_info = None - - diff = enumerate(diff, 1) - for unused_diff_line_no, line in diff: - if encoding is not None: - line = line.decode(encoding) - - # check for a git file rename - is_diff_git_header = ( - RE_DIFF_GIT_HEADER.match(line) - or RE_DIFF_GIT_HEADER_URI_LIKE.match(line) - or RE_DIFF_GIT_HEADER_NO_PREFIX.match(line) - ) - if is_diff_git_header: - patch_info = PatchInfo() - source_file = is_diff_git_header.group("source") - target_file = is_diff_git_header.group("target") - current_file = PatchedFile( - patch_info, source_file, target_file, None, None - ) - self.append(current_file) - patch_info.append(line) - continue - - # check for a git new file - is_diff_git_new_file = RE_DIFF_GIT_NEW_FILE.match(line) - if is_diff_git_new_file: - if current_file is None or patch_info is None: - raise UnidiffParseError(f"Unexpected new file found: {line}") - current_file.source_file = DEV_NULL - patch_info.append(line) - continue - - # check for a git deleted file - is_diff_git_deleted_file = RE_DIFF_GIT_DELETED_FILE.match(line) - if is_diff_git_deleted_file: - if current_file is None or patch_info is None: - raise UnidiffParseError(f"Unexpected deleted file found: {line}") - current_file.target_file = DEV_NULL - patch_info.append(line) - continue - - # check for source file header - is_source_filename = RE_SOURCE_FILENAME.match(line) - if is_source_filename: - source_file = is_source_filename.group("filename") - source_timestamp = is_source_filename.group("timestamp") - # reset current file, unless we are processing a rename - # (in that case, source files should match) - if current_file is not None and not ( - current_file.source_file == source_file - ): - current_file = None - elif current_file is not None: - current_file.source_timestamp = source_timestamp - continue - - # check for target file header - is_target_filename = RE_TARGET_FILENAME.match(line) - if is_target_filename: - target_file = is_target_filename.group("filename") - target_timestamp = is_target_filename.group("timestamp") - if current_file is not None and not ( - current_file.target_file == target_file - ): - raise UnidiffParseError(f"Target without source: {line}") - if current_file is None: - # add current file to PatchSet - current_file = PatchedFile( - patch_info, - source_file, - target_file, - source_timestamp, - target_timestamp, - ) - self.append(current_file) - patch_info = None - else: - current_file.target_timestamp = target_timestamp - continue - - # check for hunk header - is_hunk_header = RE_HUNK_HEADER.match(line) - if is_hunk_header: - patch_info = None - if current_file is None: - raise UnidiffParseError(f"Unexpected hunk found: {line}") - current_file._parse_hunk(line, diff, encoding, metadata_only) - continue - - # check for no newline marker - is_no_newline = RE_NO_NEWLINE_MARKER.match(line) - if is_no_newline: - if current_file is None: - raise UnidiffParseError(f"Unexpected marker: {line}") - current_file._add_no_newline_marker_to_last_hunk() - continue - - # sometimes hunks can be followed by empty lines - if line == "\n" and current_file is not None: - current_file._append_trailing_empty_line() - continue - - # if nothing has matched above then this line is a patch info - if patch_info is None: - current_file = None - patch_info = PatchInfo() - - is_binary_diff = RE_BINARY_DIFF.match(line) - if is_binary_diff: - source_file = is_binary_diff.group("source_filename") - target_file = is_binary_diff.group("target_filename") - patch_info.append(line) - if current_file is not None: - current_file.is_binary_file = True - else: - current_file = PatchedFile( - patch_info, source_file, target_file, is_binary_file=True - ) - self.append(current_file) - patch_info = None - current_file = None - continue - - if line == "GIT binary patch\n": - current_file.is_binary_file = True - patch_info = None - current_file = None - continue - - patch_info.append(line) - - @classmethod - def from_filename( - cls, filename, encoding=DEFAULT_ENCODING, errors=None, newline=None - ): - # type: (str, str, Optional[str]) -> PatchSet - """Return a PatchSet instance given a diff filename.""" - with open_file( - filename, "r", encoding=encoding, errors=errors, newline=newline - ) as f: - instance = cls(f) - return instance - - @staticmethod - def _convert_string(data, encoding=None, errors="strict"): - # type: (Union[str, bytes], str, str) -> StringIO - if encoding is not None: - # if encoding is given, assume bytes and decode - data = unicode(data, encoding=encoding, errors=errors) - return StringIO(data) - - @classmethod - def from_string(cls, data, encoding=None, errors="strict"): - # type: (str, str, Optional[str]) -> PatchSet - """Return a PatchSet instance given a diff string.""" - return cls(cls._convert_string(data, encoding, errors)) - - @property - def added_files(self): - # type: () -> list[PatchedFile] - """Return patch added files as a list.""" - return [f for f in self if f.is_added_file] - - @property - def removed_files(self): - # type: () -> list[PatchedFile] - """Return patch removed files as a list.""" - return [f for f in self if f.is_removed_file] - - @property - def modified_files(self): - # type: () -> list[PatchedFile] - """Return patch modified files as a list.""" - return [f for f in self if f.is_modified_file] - - @property - def added(self): - # type: () -> int - """Return the patch total added lines.""" - return sum([f.added for f in self]) - - @property - def removed(self): - # type: () -> int - """Return the patch total removed lines.""" - return sum([f.removed for f in self]) diff --git a/scanpipe/pipes/unidiff/patch.py.ABOUT b/scanpipe/pipes/unidiff/patch.py.ABOUT deleted file mode 100644 index 75c3bb8ade..0000000000 --- a/scanpipe/pipes/unidiff/patch.py.ABOUT +++ /dev/null @@ -1,13 +0,0 @@ -about_resource: patch.py constants.py errors.py -name: patch -version: 0.7.5 -download_url: https://github.com/matiasb/python-unidiff/archive/refs/tags/v0.7.5.zip -description: Simple Python library to parse and interact with unified diff data. -homepage_url: https://github.com/matiasb/python-unidiff -license_expression: mit -attribute: yes -package_url: pkg:pypi/unidiff@0.7.5 -licenses: - - key: mit - name: MIT License - file: patch.py.LICENSE \ No newline at end of file diff --git a/scanpipe/pipes/unidiff/patch.py.LICENSE b/scanpipe/pipes/unidiff/patch.py.LICENSE deleted file mode 100644 index ca2c04c202..0000000000 --- a/scanpipe/pipes/unidiff/patch.py.LICENSE +++ /dev/null @@ -1,20 +0,0 @@ -The MIT License (MIT) -Copyright (c) 2012 Matias Bordese - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, -EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF -MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. -IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, -DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR -OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE -OR OTHER DEALINGS IN THE SOFTWARE. \ No newline at end of file diff --git a/scanpipe/tests/data/reachability/app.py b/scanpipe/tests/data/reachability/app.py index b8c9eff5e0..32eebc2fa1 100644 --- a/scanpipe/tests/data/reachability/app.py +++ b/scanpipe/tests/data/reachability/app.py @@ -1,5 +1,7 @@ import os +debug = False + class ReportGenerator: """A dummy class to test AST class method parsing.""" diff --git a/scanpipe/tests/data/reachability/diff-app.patch b/scanpipe/tests/data/reachability/diff-app.patch deleted file mode 100644 index ccb86953a8..0000000000 --- a/scanpipe/tests/data/reachability/diff-app.patch +++ /dev/null @@ -1,39 +0,0 @@ -From 8f7b1c3d9a4e2b6f5d8c1a2e3f4b5c6d7e8f9a0b Mon Sep 17 00:00:00 2001 -From: Security Team -Date: Tue, 2 Jun 2026 10:00:00 +0000 -Subject: [PATCH] Fix path traversal vulnerability in report generator - -- Validates that target paths stay within the designated base_dir. -- Catches ValueError on invalid path resolution. ---- - app.py | 12 +++++++++--- - 1 file changed, 9 insertions(+), 3 deletions(-) - -diff --git a/app.py b/app.py -index a1b2c3d..e4f5g6h 100644 ---- a/app.py -+++ b/app.py -@@ -15,13 +15,19 @@ def serve_report(request_payload): - # 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) -+ # 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" - -- target_path = build_file_path(requested_file) -+ 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}" \ No newline at end of file diff --git a/scanpipe/tests/data/reachability/fixed-app.py b/scanpipe/tests/data/reachability/fixed-app.py index ca5a6f4c8b..30470b6cda 100644 --- a/scanpipe/tests/data/reachability/fixed-app.py +++ b/scanpipe/tests/data/reachability/fixed-app.py @@ -1,5 +1,7 @@ import os +debug = True + class ReportGenerator: """A dummy class to test AST class method parsing.""" diff --git a/scanpipe/tests/data/reachability/vuln-app.py b/scanpipe/tests/data/reachability/vuln-app.py index b8c9eff5e0..32eebc2fa1 100644 --- a/scanpipe/tests/data/reachability/vuln-app.py +++ b/scanpipe/tests/data/reachability/vuln-app.py @@ -1,5 +1,7 @@ import os +debug = False + class ReportGenerator: """A dummy class to test AST class method parsing.""" From a71df306a655f1e183d7f613c7258914b915d96c Mon Sep 17 00:00:00 2001 From: ziad hany Date: Mon, 10 Aug 2026 13:44:21 +0300 Subject: [PATCH 09/13] Add a test for java Add end-to-end test for Symbol Reachability pipeline Signed-off-by: ziad hany --- scanpipe/tests/data/reachability/app.py | 37 ---------------- scanpipe/tests/data/reachability/fixed-app.py | 43 ------------------- scanpipe/tests/data/reachability/vuln-app.py | 37 ---------------- 3 files changed, 117 deletions(-) delete mode 100644 scanpipe/tests/data/reachability/app.py delete mode 100644 scanpipe/tests/data/reachability/fixed-app.py delete mode 100644 scanpipe/tests/data/reachability/vuln-app.py diff --git a/scanpipe/tests/data/reachability/app.py b/scanpipe/tests/data/reachability/app.py deleted file mode 100644 index 32eebc2fa1..0000000000 --- a/scanpipe/tests/data/reachability/app.py +++ /dev/null @@ -1,37 +0,0 @@ -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 deleted file mode 100644 index 30470b6cda..0000000000 --- a/scanpipe/tests/data/reachability/fixed-app.py +++ /dev/null @@ -1,43 +0,0 @@ -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 deleted file mode 100644 index 32eebc2fa1..0000000000 --- a/scanpipe/tests/data/reachability/vuln-app.py +++ /dev/null @@ -1,37 +0,0 @@ -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." From 35c403bf2a33cdca5a773f140299cd90df27a581 Mon Sep 17 00:00:00 2001 From: ziad hany Date: Fri, 14 Aug 2026 06:11:00 +0300 Subject: [PATCH 10/13] Remove type hints Signed-off-by: ziad hany --- scanpipe/pipes/reachability.py | 1 - 1 file changed, 1 deletion(-) diff --git a/scanpipe/pipes/reachability.py b/scanpipe/pipes/reachability.py index 844324bb65..471fb566a6 100644 --- a/scanpipe/pipes/reachability.py +++ b/scanpipe/pipes/reachability.py @@ -27,7 +27,6 @@ from enum import Enum from pathlib import Path -from git import Repo from git.diff import NULL_TREE from typecode import get_type From c9e2f7c277927c3f060e6a4328665e9266d05806 Mon Sep 17 00:00:00 2001 From: ziad hany Date: Tue, 25 Aug 2026 02:41:59 +0300 Subject: [PATCH 11/13] Expose reachability in API for package and dependency Signed-off-by: ziad hany --- .../pipelines/analyze_symbols_reachability.py | 11 +++ scanpipe/pipes/reachability.py | 67 ++++++++++++++- scanpipe/tests/test_api.py | 83 +++++++++++++++++++ 3 files changed, 157 insertions(+), 4 deletions(-) diff --git a/scanpipe/pipelines/analyze_symbols_reachability.py b/scanpipe/pipelines/analyze_symbols_reachability.py index 6084612316..2c747c405b 100644 --- a/scanpipe/pipelines/analyze_symbols_reachability.py +++ b/scanpipe/pipelines/analyze_symbols_reachability.py @@ -57,6 +57,7 @@ def steps(cls): 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): @@ -101,3 +102,13 @@ def generate_advisory_reachability_report(self): 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.advisory_map + ) diff --git a/scanpipe/pipes/reachability.py b/scanpipe/pipes/reachability.py index 471fb566a6..4199bbd056 100644 --- a/scanpipe/pipes/reachability.py +++ b/scanpipe/pipes/reachability.py @@ -31,6 +31,8 @@ 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 @@ -783,7 +785,7 @@ def generate_advisory_reachability_report(project, patches, candidate_resources) ReachabilityStatus.NOT_REACHABLE.value: 1, } - advisory_reachability_report = { + advisories_reachability_report = { "purl": project.purl, "advisories": [], } @@ -798,7 +800,7 @@ def generate_advisory_reachability_report(project, patches, candidate_resources) "details": [], } advisory_map[adv_uid] = adv_data - advisory_reachability_report["advisories"].append(adv_data) + advisories_reachability_report["advisories"].append(adv_data) for resource in candidate_resources: for report in resource.extra_data.get("symbols_reachability", []): @@ -816,7 +818,7 @@ def generate_advisory_reachability_report(project, patches, candidate_resources) "details": [], } advisory_map[adv_uid] = adv_data - advisory_reachability_report["advisories"].append(adv_data) + advisories_reachability_report["advisories"].append(adv_data) tool_details = { "resource_path": resource.path, @@ -838,4 +840,61 @@ def generate_advisory_reachability_report(project, patches, candidate_resources) reachability_output_path = project.get_output_file_path("reachability", "json") with open(reachability_output_path, "w") as f: - json.dump(advisory_reachability_report, f, indent=2) + 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_list = advisory_report.get("advisories", []) + if not advisories_list: + return + + advisory_map = {adv["advisory_uid"]: adv for adv in advisories_list} + + unsaved_packages = [] + for package in project.discoveredpackages.all(): + vulns = package.affected_by_vulnerabilities or [] + if inject_reachability_data(vulns, advisory_map): + unsaved_packages.append(package) + + if unsaved_packages: + DiscoveredPackage.objects.bulk_update( + objs=unsaved_packages, + fields=["affected_by_vulnerabilities"], + batch_size=10, + ) + + unsaved_deps = [] + for dep in project.discovereddependencies.all(): + vulns = dep.affected_by_vulnerabilities or [] + if inject_reachability_data(vulns, advisory_map): + unsaved_deps.append(dep) + + if unsaved_deps: + DiscoveredDependency.objects.bulk_update( + objs=unsaved_deps, + fields=["affected_by_vulnerabilities"], + batch_size=10, + ) 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]) From 5d9ba9bcbc5efa873568c6a92d26097e17fc32da Mon Sep 17 00:00:00 2001 From: ziad hany Date: Tue, 25 Aug 2026 14:39:49 +0300 Subject: [PATCH 12/13] Simplify the apply_reachability_to_packages_and_dependencies function Signed-off-by: ziad hany --- scanpipe/pipes/reachability.py | 51 +++++++++++++++------------------- 1 file changed, 22 insertions(+), 29 deletions(-) diff --git a/scanpipe/pipes/reachability.py b/scanpipe/pipes/reachability.py index 4199bbd056..c71981cb2f 100644 --- a/scanpipe/pipes/reachability.py +++ b/scanpipe/pipes/reachability.py @@ -867,34 +867,27 @@ 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_list = advisory_report.get("advisories", []) - if not advisories_list: + advisories = advisory_report.get("advisories", []) + if not advisories: return - advisory_map = {adv["advisory_uid"]: adv for adv in advisories_list} - - unsaved_packages = [] - for package in project.discoveredpackages.all(): - vulns = package.affected_by_vulnerabilities or [] - if inject_reachability_data(vulns, advisory_map): - unsaved_packages.append(package) - - if unsaved_packages: - DiscoveredPackage.objects.bulk_update( - objs=unsaved_packages, - fields=["affected_by_vulnerabilities"], - batch_size=10, - ) - - unsaved_deps = [] - for dep in project.discovereddependencies.all(): - vulns = dep.affected_by_vulnerabilities or [] - if inject_reachability_data(vulns, advisory_map): - unsaved_deps.append(dep) - - if unsaved_deps: - DiscoveredDependency.objects.bulk_update( - objs=unsaved_deps, - fields=["affected_by_vulnerabilities"], - batch_size=10, - ) + 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, + ) From b8ece711e236111eb341d5b7010b48d53df7f1ed Mon Sep 17 00:00:00 2001 From: ziad hany Date: Sat, 29 Aug 2026 00:43:39 +0300 Subject: [PATCH 13/13] Resolve merge conflict Signed-off-by: ziad hany --- scanpipe/pipelines/analyze_symbols_reachability.py | 12 +++++++----- scanpipe/pipes/reachability.py | 1 + 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/scanpipe/pipelines/analyze_symbols_reachability.py b/scanpipe/pipelines/analyze_symbols_reachability.py index 2c747c405b..4eaf4116e1 100644 --- a/scanpipe/pipelines/analyze_symbols_reachability.py +++ b/scanpipe/pipelines/analyze_symbols_reachability.py @@ -97,10 +97,12 @@ def collect_and_match_resources(self): def generate_advisory_reachability_report(self): """Generate a reachability report summarizing status by advisory.""" - reachability.generate_advisory_reachability_report( - project=self.project, - patches=self.patches, - candidate_resources=self.candidate_resources, + 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): @@ -110,5 +112,5 @@ def apply_reachability_to_packages_and_dependencies(self): in their affected_by_vulnerabilities JSON field. """ reachability.apply_reachability_to_packages_and_dependencies( - project=self.project, advisory_report=self.advisory_map + project=self.project, advisory_report=self.advisories_reachability_report ) diff --git a/scanpipe/pipes/reachability.py b/scanpipe/pipes/reachability.py index c71981cb2f..ee0dc5993a 100644 --- a/scanpipe/pipes/reachability.py +++ b/scanpipe/pipes/reachability.py @@ -27,6 +27,7 @@ from enum import Enum from pathlib import Path +from git import Repo from git.diff import NULL_TREE from typecode import get_type